diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 744125da..a6a8906c 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -26,6 +26,7 @@ on:
       - 'packaging/**'
       - '.github/workflows/mcp-packaging.yml'
       - '.github/workflows/mcp-registry-publish.yml'
+      - '.github/workflows/contributors.yml'
   pull_request:
     branches:
       - main
@@ -49,6 +50,7 @@ on:
       - 'packaging/**'
       - '.github/workflows/mcp-packaging.yml'
       - '.github/workflows/mcp-registry-publish.yml'
+      - '.github/workflows/contributors.yml'
 env:
   GOTESTCMD: "go test -timeout 1200s --tags \"sqlite_stackql\" -v ./..."
   TESTSCRIPT: "test/deprecated/python/main.py"
diff --git a/.github/workflows/contributors.yml b/.github/workflows/contributors.yml
new file mode 100644
index 00000000..33ca718d
--- /dev/null
+++ b/.github/workflows/contributors.yml
@@ -0,0 +1,80 @@
+name: contributors
+
+# Regenerates internal/stackql/contributors/contributors.csv, the leaderboard
+# behind SHOW CONTRIBUTORS (issue #320), using stackql itself: human
+# contributors who have starred the repo, contributions descending. Opens a
+# PR so every change to the credit list is reviewed. PRs opened with the
+# workflow token do not trigger CI; close and reopen the PR to run it.
+
+on:
+  schedule:
+    - cron: '17 3 1 * *'
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  pull-requests: write
+
+env:
+  CSV_PATH: internal/stackql/contributors/contributors.csv
+  # Machine accounts operating as regular user accounts (PAT-driven CI
+  # identities, AI coding agents with user accounts). GitHub Apps are
+  # excluded generically by the type and [bot] filters in the query.
+  DENYLIST: "'claude'"
+  # An API failure or truncated result must not replace a healthy list.
+  MIN_ROWS: 10
+
+jobs:
+  contributors:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: stackql/setup-stackql@v2.3.1
+
+      - name: Query contributors
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          stackql exec "REGISTRY PULL github;"
+          stackql exec \
+            --auth='{"github":{"type":"bearer","credentialsenvvar":"GITHUB_TOKEN"}}' \
+            --http.response.pageLimit=-1 --http.response.maxResults=100 \
+            --output csv \
+            "SELECT c.login AS contributor, c.contributions
+             FROM github.repos.contributors c
+             INNER JOIN github.activity.repo_stargazers s ON s.login = c.login
+             WHERE c.owner = 'stackql' AND c.repo = 'stackql'
+             AND s.owner = 'stackql' AND s.repo = 'stackql'
+             AND c.type = 'User'
+             AND c.login NOT LIKE '%[bot]'
+             AND c.login NOT IN (${DENYLIST})
+             ORDER BY c.contributions DESC, c.login ASC" > "${CSV_PATH}"
+
+      - name: Guard contributor count
+        run: |
+          count=$(tail -n +2 "${CSV_PATH}" | wc -l)
+          if [ "${count}" -le "${MIN_ROWS}" ]; then
+            echo "::error::contributor query returned ${count} rows (<= ${MIN_ROWS}), refusing to update ${CSV_PATH}"
+            exit 1
+          fi
+          echo "${count} contributors"
+
+      - name: Open PR on change
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          if git diff --quiet -- "${CSV_PATH}"; then
+            echo "leaderboard unchanged"
+            exit 0
+          fi
+          branch="contributors/$(date -u +%Y%m%d)"
+          git config user.name "github-actions[bot]"
+          git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
+          git checkout -b "${branch}"
+          git add "${CSV_PATH}"
+          git commit -m "contributors: refresh leaderboard"
+          git push -u origin "${branch}"
+          gh pr create --base main --head "${branch}" \
+            --title "contributors: refresh leaderboard" \
+            --body "Regenerated by the contributors workflow. Human contributors who have starred the repo, contributions descending."
diff --git a/cicd/version.txt b/cicd/version.txt
index c783f880..16cce635 100644
--- a/cicd/version.txt
+++ b/cicd/version.txt
@@ -1,2 +1,2 @@
 MajorVersion=0
-MinorVersion=10
+MinorVersion=11
diff --git a/go.mod b/go.mod
index dd815fb5..64316177 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e
 	github.com/getkin/kin-openapi v0.88.0
 	github.com/google/go-jsonnet v0.17.0
+	github.com/google/uuid v1.6.0
 	github.com/jackc/pgtype v1.10.0
 	github.com/jackc/pgx/v5 v5.0.4
 	github.com/lib/pq v1.10.4
@@ -93,7 +94,6 @@ require (
 	github.com/google/flatbuffers v24.12.23+incompatible // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/jsonschema-go v0.4.3 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/mux v1.8.0 // indirect
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
diff --git a/internal/stackql/cmd/root.go b/internal/stackql/cmd/root.go
index a5ded201..5480d274 100644
--- a/internal/stackql/cmd/root.go
+++ b/internal/stackql/cmd/root.go
@@ -212,7 +212,7 @@ func init() {
 	rootCmd.PersistentFlags().BoolVarP(&runtimeCtx.VerboseFlag, dto.VerboseFlagKey, "v", false, "Verbose flag")
 	rootCmd.PersistentFlags().BoolVar(&runtimeCtx.DryRunFlag, dto.DryRunFlagKey, false, "dryrun flag; preprocessor only will run and output returned")
 	rootCmd.PersistentFlags().BoolVarP(&runtimeCtx.CSVHeadersDisable, dto.CSVHeadersDisableKey, "H", false, "Disable CSV headers flag")
-	rootCmd.PersistentFlags().StringVarP(&runtimeCtx.OutputFormat, dto.OutputFormatKey, "o", "table", "Output format, must be (json | jsonl | table | csv)")
+	rootCmd.PersistentFlags().StringVarP(&runtimeCtx.OutputFormat, dto.OutputFormatKey, "o", "table", "Output format, must be (json | jsonl | table | csv | text | otel)")
 	rootCmd.PersistentFlags().StringVarP(&runtimeCtx.OutfilePath, dto.OutfilePathKey, "f", "stdout", "Output file into which results are written")
 	rootCmd.PersistentFlags().StringVarP(&runtimeCtx.InfilePath, dto.InfilePathKey, "i", "stdin", "Input file from which queries are read")
 	rootCmd.PersistentFlags().StringVarP(&runtimeCtx.TemplateCtxFilePath, dto.TemplateCtxFilePathKey, "q", "", "Context file for templating")
diff --git a/internal/stackql/contributors/contributors.csv b/internal/stackql/contributors/contributors.csv
new file mode 100644
index 00000000..35221a38
--- /dev/null
+++ b/internal/stackql/contributors/contributors.csv
@@ -0,0 +1,18 @@
+contributor,contributions
+general-kroll-4-life,627
+jeffreyaven,50
+Iqbalabdi,4
+Sneha41sb,4
+jthegedus,4
+derek10cloud,3
+nirmalpatel008,2
+Racso-3141,1
+adityagiri3600,1
+datwiz,1
+kieranrimmer,1
+ryuichi-maeda,1
+ryukinoz,1
+sanketmp,1
+soysaucewaso,1
+tomekz,1
+yunchengyang515,1
diff --git a/internal/stackql/contributors/contributors.go b/internal/stackql/contributors/contributors.go
new file mode 100644
index 00000000..3182134e
--- /dev/null
+++ b/internal/stackql/contributors/contributors.go
@@ -0,0 +1,65 @@
+// Package contributors credits the humans behind stackql: the vendored
+// leaderboard behind SHOW CONTRIBUTORS (issue #320). contributors.csv is
+// regenerated by the contributors workflow and compiled in, so the build
+// stays hermetic and each release freezes its credits.
+package contributors
+
+import (
+	_ "embed"
+	"encoding/csv"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+//go:embed contributors.csv
+var contributorsCSV string //nolint:gochecknoglobals // embed target
+
+// csvFields is the column count of contributors.csv: contributor,contributions.
+const csvFields = 2
+
+// Contributor is one leaderboard entry.
+type Contributor interface {
+	GetLogin() string
+	GetContributions() int
+}
+
+type contributor struct {
+	login         string
+	contributions int
+}
+
+func (c contributor) GetLogin() string      { return c.login }
+func (c contributor) GetContributions() int { return c.contributions }
+
+//nolint:gochecknoglobals // parsed once
+var (
+	parseOnce sync.Once
+	parsed    []Contributor
+)
+
+// List returns the leaderboard in vendored order (contributions descending).
+func List() []Contributor {
+	parseOnce.Do(func() { parsed = parse(contributorsCSV) })
+	return parsed
+}
+
+// parse skips the header and any malformed line.
+func parse(raw string) []Contributor {
+	records, err := csv.NewReader(strings.NewReader(raw)).ReadAll()
+	if err != nil || len(records) == 0 {
+		return nil
+	}
+	rv := make([]Contributor, 0, len(records)-1)
+	for _, rec := range records[1:] {
+		if len(rec) != csvFields {
+			continue
+		}
+		n, convErr := strconv.Atoi(strings.TrimSpace(rec[1]))
+		if convErr != nil || strings.TrimSpace(rec[0]) == "" {
+			continue
+		}
+		rv = append(rv, contributor{login: strings.TrimSpace(rec[0]), contributions: n})
+	}
+	return rv
+}
diff --git a/internal/stackql/contributors/contributors_test.go b/internal/stackql/contributors/contributors_test.go
new file mode 100644
index 00000000..28d3e62b
--- /dev/null
+++ b/internal/stackql/contributors/contributors_test.go
@@ -0,0 +1,33 @@
+package contributors_test
+
+import (
+	"testing"
+
+	"github.com/stackql/stackql/internal/stackql/contributors"
+)
+
+// The embedded leaderboard must be non-empty, ordered by contributions
+// descending, and free of blank or automated logins.
+func TestList_EmbeddedLeaderboard(t *testing.T) {
+	list := contributors.List()
+	if len(list) == 0 {
+		t.Fatal("embedded contributors.csv parsed to an empty leaderboard")
+	}
+	for i, c := range list {
+		if c.GetLogin() == "" || c.GetContributions() <= 0 {
+			t.Fatalf("entry %d is malformed: %q %d", i, c.GetLogin(), c.GetContributions())
+		}
+		if i > 0 && c.GetContributions() > list[i-1].GetContributions() {
+			t.Fatalf("entry %d (%s) breaks descending order", i, c.GetLogin())
+		}
+	}
+	if list[0].GetLogin() != "general-kroll-4-life" {
+		t.Fatalf("expected the project founder on top, got %q", list[0].GetLogin())
+	}
+}
+
+func TestList_IsStable(t *testing.T) {
+	if a, b := contributors.List(), contributors.List(); len(a) != len(b) || &a[0] != &b[0] {
+		t.Fatal("List() must parse once and return the same leaderboard")
+	}
+}
diff --git a/internal/stackql/driver/driver.go b/internal/stackql/driver/driver.go
index f7fb090d..8dc99609 100644
--- a/internal/stackql/driver/driver.go
+++ b/internal/stackql/driver/driver.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/stackql/any-sdk/pkg/logging"
 	"github.com/stackql/any-sdk/public/sqlengine"
@@ -124,6 +125,7 @@ func (dr *basicStackQLDriver) ProcessDryRun(query string) {
 func (dr *basicStackQLDriver) ProcessQuery(query string) {
 	clonedCtx := dr.handlerCtx.Clone()
 	clonedCtx.SetRawQuery(query)
+	clonedCtx.SetQueryStartTime(time.Now())
 	responses, ok := dr.processQueryOrQueries(clonedCtx)
 	if ok {
 		for _, r := range responses {
diff --git a/internal/stackql/handler/handler.go b/internal/stackql/handler/handler.go
index 0900027e..9a22e4fa 100644
--- a/internal/stackql/handler/handler.go
+++ b/internal/stackql/handler/handler.go
@@ -9,6 +9,7 @@ import (
 	"path"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/stackql/any-sdk/pkg/constants"
 	"github.com/stackql/any-sdk/pkg/db/sqlcontrol"
@@ -57,6 +58,7 @@ type HandlerContext interface { //nolint:revive // don't mind stuttering this on
 	//
 	GetRawQuery() string
 	GetQuery() string
+	GetQueryStartTime() time.Time
 	GetRuntimeContext() dto.RuntimeCtx
 	GetProviders() map[string]provider.IProvider
 	GetControlAttributes() sqlcontrol.ControlAttributes
@@ -83,6 +85,7 @@ type HandlerContext interface { //nolint:revive // don't mind stuttering this on
 	SetCurrentProvider(string)
 	SetQuery(string)
 	SetRawQuery(string)
+	SetQueryStartTime(time.Time)
 	//
 	SetOutErrFile(io.Writer)
 	//
@@ -126,6 +129,7 @@ type standardHandlerContext struct {
 	providersMapMutex   *sync.Mutex
 	rawQuery            string
 	query               string
+	queryStartTime      time.Time
 	runtimeContext      dto.RuntimeCtx
 	providers           map[string]provider.IProvider
 	controlAttributes   sqlcontrol.ControlAttributes
@@ -214,6 +218,10 @@ func (hc *standardHandlerContext) SetRawQuery(rq string) {
 	hc.rawQuery = rq
 }
 
+func (hc *standardHandlerContext) SetQueryStartTime(t time.Time) {
+	hc.queryStartTime = t
+}
+
 func (hc *standardHandlerContext) SetQuery(q string) {
 	hc.query = q
 }
@@ -223,6 +231,7 @@ func (hc *standardHandlerContext) SetTxnCounterMgr(mgr txncounter.Manager) {
 }
 
 func (hc *standardHandlerContext) GetRawQuery() string                         { return hc.rawQuery }
+func (hc *standardHandlerContext) GetQueryStartTime() time.Time                { return hc.queryStartTime }
 func (hc *standardHandlerContext) GetQuery() string                            { return hc.query }
 func (hc *standardHandlerContext) GetRuntimeContext() dto.RuntimeCtx           { return hc.runtimeContext }
 func (hc *standardHandlerContext) GetProviders() map[string]provider.IProvider { return hc.providers }
@@ -568,6 +577,7 @@ func (hc *standardHandlerContext) Clone() HandlerContext {
 		providersMapMutex:    hc.providersMapMutex,
 		drmConfig:            hc.drmConfig,
 		rawQuery:             hc.rawQuery,
+		queryStartTime:       hc.queryStartTime,
 		runtimeContext:       hc.runtimeContext,
 		currentProvider:      hc.currentProvider,
 		providers:            hc.providers,
diff --git a/internal/stackql/internal_data_transfer/internaldto/output_ctx.go b/internal/stackql/internal_data_transfer/internaldto/output_ctx.go
index d67832f1..2491f6d8 100644
--- a/internal/stackql/internal_data_transfer/internaldto/output_ctx.go
+++ b/internal/stackql/internal_data_transfer/internaldto/output_ctx.go
@@ -1,6 +1,8 @@
 package internaldto
 
 import (
+	"time"
+
 	"github.com/stackql/any-sdk/pkg/dto"
 	"github.com/stackql/psql-wire/pkg/sqldata"
 )
@@ -8,4 +10,8 @@ import (
 type OutputContext struct {
 	RuntimeContext dto.RuntimeCtx
 	Result         sqldata.ISQLResultStream
+	// Query and StartTime describe the submitted statement; the otel writer
+	// stamps them on every record.
+	Query     string
+	StartTime time.Time
 }
diff --git a/internal/stackql/output/otel.go b/internal/stackql/output/otel.go
new file mode 100644
index 00000000..fce4f502
--- /dev/null
+++ b/internal/stackql/output/otel.go
@@ -0,0 +1,313 @@
+package output
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stackql/psql-wire/pkg/sqldata"
+	"github.com/stackql/stackql-parser/go/vt/sqlparser"
+	"github.com/stackql/stackql/internal/stackql/buildinfo"
+	"github.com/stackql/stackql/pkg/sink"
+)
+
+// The emitted attribute set is a versioned interface (issue #738): bump
+// otelAttributeSchemaVersion when it changes and update the schema test.
+const (
+	otelFormatStr              = "otel"
+	otelAttributeSchemaVersion = "1.0.0"
+	otelScopeName              = "github.com/stackql/stackql/internal/stackql/output"
+	otelServiceName            = "stackql"
+	otelTraceParentEnv         = "TRACEPARENT"
+	otelServiceNameEnv         = "OTEL_SERVICE_NAME"
+	otelResourceAttrsEnv       = "OTEL_RESOURCE_ATTRIBUTES"
+	otelTraceIDBytes           = 16
+	otelSpanIDBytes            = 8
+	otelTraceParentParts       = 4
+)
+
+// otelInvocationTraceID is shared by every statement of one process,
+// honouring a TRACEPARENT supplied by the environment.
+//
+//nolint:gochecknoglobals // per-process trace identity
+var otelInvocationTraceID = sync.OnceValue(func() string {
+	parts := strings.Split(os.Getenv(otelTraceParentEnv), "-")
+	if len(parts) == otelTraceParentParts && len(parts[1]) == 2*otelTraceIDBytes {
+		return parts[1]
+	}
+	return sink.OTelRandomID(otelTraceIDBytes)
+})
+
+// OTelWriter emits one OTLP/JSON LogsData per result row plus a completion
+// record per statement, so a result set can be shipped as a timestamped
+// inventory snapshot. Every record of a statement shares the snapshot
+// instant, snapshot id and span id.
+type OTelWriter struct {
+	errWriter  io.Writer
+	encoder    sink.Sink
+	startTime  time.Time
+	snapshotID string
+	spanID     string
+	query      string
+	queryHash  string
+	statement  otelStatementContext
+}
+
+// otelStatementContext is what a lightweight parse of the statement yields:
+// the primary table and any cloud scoping parameters.
+type otelStatementContext struct {
+	provider string
+	service  string
+	resource string
+	cloud    []sink.OTelAttribute
+}
+
+// otelRecordPayload adapts a prepared record to the sink decorator.
+type otelRecordPayload struct {
+	record sink.OTelLogRecord
+}
+
+func (p otelRecordPayload) OTelLogRecords() []sink.OTelLogRecord {
+	return []sink.OTelLogRecord{p.record}
+}
+
+// writerSink is the io.Writer-backed sink the decorator writes through: one
+// compact JSON document per line, flushed per record like the jsonl writer.
+type writerSink struct {
+	writer io.Writer
+}
+
+func (s writerSink) Record(_ context.Context, payload any) error {
+	b, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	if _, err = s.writer.Write(append(b, '\n')); err != nil {
+		return err
+	}
+	if flusher, ok := s.writer.(writeFlusher); ok {
+		return flusher.Flush()
+	}
+	return nil
+}
+
+func (writerSink) Close() error { return nil }
+
+// NewOTelWriter returns the writer for one statement: query is the submitted
+// text and startTime the snapshot instant stamped on every record.
+func NewOTelWriter(writer io.Writer, errWriter io.Writer, query string, startTime time.Time) IOutputWriter {
+	if startTime.IsZero() {
+		startTime = time.Now()
+	}
+	statement := newOTelStatementContext(query)
+	queryHash := sha256.Sum256([]byte(strings.TrimSpace(query)))
+	return &OTelWriter{
+		errWriter:  errWriter,
+		encoder:    sink.NewOTelSink(writerSink{writer: writer}, otelResource(statement), otelScope()),
+		startTime:  startTime,
+		snapshotID: uuid.NewString(),
+		spanID:     sink.OTelRandomID(otelSpanIDBytes),
+		query:      query,
+		queryHash:  hex.EncodeToString(queryHash[:]),
+		statement:  statement,
+	}
+}
+
+func otelScope() sink.OTelScope {
+	return sink.OTelScope{Name: otelScopeName, Version: otelAttributeSchemaVersion}
+}
+
+// otelResource is service.*, the derivable cloud.* conventions, then the
+// standard OTEL_RESOURCE_ATTRIBUTES pairs.
+func otelResource(statement otelStatementContext) sink.OTelResource {
+	serviceName := os.Getenv(otelServiceNameEnv)
+	if serviceName == "" {
+		serviceName = otelServiceName
+	}
+	attrs := append([]sink.OTelAttribute{}, statement.cloud...)
+	for _, pair := range strings.Split(os.Getenv(otelResourceAttrsEnv), ",") {
+		key, value, found := strings.Cut(pair, "=")
+		if !found || strings.TrimSpace(key) == "" {
+			continue
+		}
+		if unescaped, err := url.PathUnescape(value); err == nil {
+			value = unescaped
+		}
+		attrs = append(attrs, sink.OTelString(strings.TrimSpace(key), strings.TrimSpace(value)))
+	}
+	return sink.OTelResource{
+		ServiceName:    serviceName,
+		ServiceVersion: buildinfo.Get().GetSemVersion(),
+		Attributes:     attrs,
+	}
+}
+
+// newOTelStatementContext parses a single-table SELECT for its
+// provider.service.resource and the equality parameters that map onto the
+// cloud resource conventions; anything more complex yields nothing.
+func newOTelStatementContext(query string) otelStatementContext {
+	var rv otelStatementContext
+	stmt, err := sqlparser.Parse(query)
+	if err != nil {
+		return rv
+	}
+	sel, ok := stmt.(*sqlparser.Select)
+	if !ok || len(sel.From) != 1 {
+		return rv
+	}
+	aliased, ok := sel.From[0].(*sqlparser.AliasedTableExpr)
+	if !ok {
+		return rv
+	}
+	tableName, ok := aliased.Expr.(sqlparser.TableName)
+	if !ok || tableName.QualifierSecond.IsEmpty() {
+		return rv
+	}
+	rv.provider = tableName.QualifierSecond.GetRawVal()
+	rv.service = tableName.Qualifier.GetRawVal()
+	rv.resource = tableName.Name.GetRawVal()
+	params := map[string]string{}
+	if sel.Where != nil {
+		otelCollectEqualities(sel.Where.Expr, params)
+	}
+	rv.cloud = otelCloudAttributes(rv.provider, params)
+	return rv
+}
+
+// otelCollectEqualities gathers `column = 'literal'` terms of an AND chain.
+func otelCollectEqualities(expr sqlparser.Expr, out map[string]string) {
+	switch node := expr.(type) {
+	case *sqlparser.AndExpr:
+		otelCollectEqualities(node.Left, out)
+		otelCollectEqualities(node.Right, out)
+	case *sqlparser.ComparisonExpr:
+		col, isCol := node.Left.(*sqlparser.ColName)
+		val, isVal := node.Right.(*sqlparser.SQLVal)
+		if node.Operator == sqlparser.EqualStr && isCol && isVal && val.Type == sqlparser.StrVal {
+			out[strings.ToLower(col.Name.GetRawVal())] = string(val.Val)
+		}
+	}
+}
+
+func otelCloudAttributes(providerName string, params map[string]string) []sink.OTelAttribute {
+	var attrs []sink.OTelAttribute
+	switch {
+	case strings.HasPrefix(providerName, "google"):
+		attrs = append(attrs, sink.OTelString("cloud.provider", "gcp"))
+	case strings.HasPrefix(providerName, "aws"):
+		attrs = append(attrs, sink.OTelString("cloud.provider", "aws"))
+	case strings.HasPrefix(providerName, "azure"):
+		attrs = append(attrs, sink.OTelString("cloud.provider", "azure"))
+	case providerName == "oci":
+		attrs = append(attrs, sink.OTelString("cloud.provider", "oracle_cloud"))
+	}
+	for param, key := range map[string]string{
+		"project":        "cloud.account.id",
+		"subscriptionid": "cloud.account.id",
+		"region":         "cloud.region",
+		"zone":           "cloud.availability_zone",
+	} {
+		if v := params[param]; v != "" {
+			attrs = append(attrs, sink.OTelString(key, v))
+		}
+	}
+	return attrs
+}
+
+// contextAttributes are the stackql.* attributes common to every record of
+// the statement.
+func (ow *OTelWriter) contextAttributes() []sink.OTelAttribute {
+	return []sink.OTelAttribute{
+		sink.OTelString("stackql.snapshot.id", ow.snapshotID),
+		sink.OTelString("stackql.query", ow.query),
+		sink.OTelString("stackql.query.hash", ow.queryHash),
+		sink.OTelString("stackql.provider", ow.statement.provider),
+		sink.OTelString("stackql.service", ow.statement.service),
+		sink.OTelString("stackql.resource", ow.statement.resource),
+	}
+}
+
+func (ow *OTelWriter) record(severity int, body string, attrs []sink.OTelAttribute) error {
+	return ow.encoder.Record(context.Background(), otelRecordPayload{sink.OTelLogRecord{
+		Time:        ow.startTime,
+		Severity:    severity,
+		Body:        body,
+		Attributes:  attrs,
+		TraceParent: fmt.Sprintf("00-%s-%s-01", otelInvocationTraceID(), ow.spanID),
+	}})
+}
+
+// writeRow emits one row: the full row JSON as the body, scalar columns as
+// typed attributes, nested columns as JSON strings, plus the row's index and
+// a fingerprint of the canonical row JSON so drift is one comparison.
+func (ow *OTelWriter) writeRow(row map[string]interface{}, index int) error {
+	body, err := json.Marshal(row)
+	if err != nil {
+		return err
+	}
+	attrs, _ := sink.OTelAttributesFromJSON(body)
+	fingerprint := sha256.Sum256(body)
+	attrs = append(attrs, ow.contextAttributes()...)
+	attrs = append(attrs,
+		sink.OTelInt("stackql.row.index", int64(index)),
+		sink.OTelString("stackql.row.fingerprint", "sha256:"+hex.EncodeToString(fingerprint[:])),
+	)
+	return ow.record(sink.OTelSeverityInfo, string(body), attrs)
+}
+
+// writeCompletion marks the snapshot whole; a zero-row statement still emits
+// it so an absent resource is visible as drift rather than as silence.
+func (ow *OTelWriter) writeCompletion(rows int) error {
+	body := "snapshot complete"
+	if ow.statement.provider != "" {
+		body += " " + strings.Join([]string{ow.statement.provider, ow.statement.service, ow.statement.resource}, ".")
+	}
+	attrs := append(ow.contextAttributes(),
+		sink.OTelBool("stackql.snapshot.complete", true),
+		sink.OTelInt("stackql.rows_returned", int64(rows)),
+		sink.OTelInt("stackql.duration_ms", time.Since(ow.startTime).Milliseconds()),
+	)
+	return ow.record(sink.OTelSeverityInfo, body, attrs)
+}
+
+func (ow *OTelWriter) Write(res sqldata.ISQLResultStream) error {
+	rows := 0
+	for {
+		r, err := res.Read()
+		if err != nil && !errors.Is(err, io.EOF) {
+			return err
+		}
+		if r != nil {
+			for _, row := range r.ToArr() {
+				if rowErr := ow.writeRow(row, rows); rowErr != nil {
+					return rowErr
+				}
+				rows++
+			}
+		}
+		if err != nil {
+			return ow.writeCompletion(rows)
+		}
+	}
+}
+
+func (ow *OTelWriter) WriteError(err error, errorPresentation string) error {
+	if errorPresentation == stderrPressentationStr {
+		return writeStderrError(ow.errWriter, err)
+	}
+	attrs := append(ow.contextAttributes(),
+		sink.OTelString("error.type", fmt.Sprintf("%T", err)),
+		sink.OTelString("error.message", err.Error()),
+	)
+	return ow.record(sink.OTelSeverityError, err.Error(), attrs)
+}
diff --git a/internal/stackql/output/otel_test.go b/internal/stackql/output/otel_test.go
new file mode 100644
index 00000000..c5fa7220
--- /dev/null
+++ b/internal/stackql/output/otel_test.go
@@ -0,0 +1,248 @@
+package output //nolint:testpackage // do not care
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stackql/any-sdk/pkg/dto"
+	"github.com/stackql/psql-wire/pkg/sqldata"
+	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
+	"github.com/stackql/stackql/pkg/sink"
+)
+
+const otelTestQuery = "select name, region, disabled, spec from google.compute.firewalls " +
+	"where project = 'p1' and zone = 'z1'"
+
+func otelTestStream(rows ...[]interface{}) sqldata.ISQLResultStream {
+	table := sqldata.NewSQLTable(0, "t")
+	cols := []sqldata.ISQLColumn{}
+	for i, name := range []string{"name", "region", "disabled", "spec"} {
+		cols = append(cols, sqldata.NewSQLColumn(table, name, int16(i), 0, 0, 0, "text")) //nolint:gosec // test
+	}
+	sqlRows := make([]sqldata.ISQLRow, 0, len(rows))
+	for _, r := range rows {
+		sqlRows = append(sqlRows, sqldata.NewSQLRow(r))
+	}
+	return sqldata.NewSimpleSQLResultStream(sqldata.NewSQLResult(cols, 0, 0, sqlRows))
+}
+
+// otelLines decodes one LogsData per line and returns its single record
+// with the resource and scope it was wrapped in.
+func otelLines(t *testing.T, out string) []sink.OTelResourceLogs {
+	t.Helper()
+	var rv []sink.OTelResourceLogs
+	for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
+		var data sink.OTelLogsData
+		if err := json.Unmarshal([]byte(line), &data); err != nil {
+			t.Fatalf("line is not LogsData: %v: %q", err, line)
+		}
+		if len(data.ResourceLogs) != 1 || len(data.ResourceLogs[0].ScopeLogs) != 1 ||
+			len(data.ResourceLogs[0].ScopeLogs[0].LogRecords) != 1 {
+			t.Fatalf("expected one record per line, got %q", line)
+		}
+		rv = append(rv, data.ResourceLogs[0])
+	}
+	return rv
+}
+
+func otelRecord(rl sink.OTelResourceLogs) sink.OTelWireLogRecord {
+	return rl.ScopeLogs[0].LogRecords[0]
+}
+
+func otelAttr(attrs []sink.OTelAttribute, key string) (sink.OTelAnyValue, bool) {
+	for _, a := range attrs {
+		if a.Key == key {
+			return a.Value, true
+		}
+	}
+	return sink.OTelAnyValue{}, false
+}
+
+func otelString(t *testing.T, attrs []sink.OTelAttribute, key string) string {
+	t.Helper()
+	v, ok := otelAttr(attrs, key)
+	if !ok || v.StringValue == nil {
+		t.Fatalf("missing string attribute %q in %+v", key, attrs)
+	}
+	return *v.StringValue
+}
+
+func otelKeys(attrs []sink.OTelAttribute) []string {
+	keys := make([]string, 0, len(attrs))
+	for _, a := range attrs {
+		keys = append(keys, a.Key)
+	}
+	return keys
+}
+
+// otelSchemaFixture writes two rows and returns the decoded lines.
+func otelSchemaFixture(t *testing.T) []sink.OTelResourceLogs {
+	t.Helper()
+	var out bytes.Buffer
+	start := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC)
+	w := NewOTelWriter(&out, &bytes.Buffer{}, otelTestQuery, start)
+	stream := otelTestStream(
+		[]interface{}{"fw-a", "us-east1", false, map[string]interface{}{"priority": 1000}},
+		[]interface{}{"fw-b", "us-east1", true, nil},
+	)
+	if err := w.Write(stream); err != nil {
+		t.Fatalf("Write() error = %v", err)
+	}
+	lines := otelLines(t, out.String())
+	if len(lines) != 3 {
+		t.Fatalf("expected 2 rows + completion, got %d lines", len(lines))
+	}
+	return lines
+}
+
+func TestOTelWriter_AttributeSchema(t *testing.T) {
+	lines := otelSchemaFixture(t)
+	wantRow := []string{
+		"disabled", "name", "region", "spec",
+		"stackql.snapshot.id", "stackql.query", "stackql.query.hash",
+		"stackql.provider", "stackql.service", "stackql.resource",
+		"stackql.row.index", "stackql.row.fingerprint",
+	}
+	if got := otelKeys(otelRecord(lines[0]).Attributes); !reflect.DeepEqual(got, wantRow) {
+		t.Fatalf("row attribute schema drift:\n got %v\nwant %v", got, wantRow)
+	}
+	wantCompletion := []string{
+		"stackql.snapshot.id", "stackql.query", "stackql.query.hash",
+		"stackql.provider", "stackql.service", "stackql.resource",
+		"stackql.snapshot.complete", "stackql.rows_returned", "stackql.duration_ms",
+	}
+	if got := otelKeys(otelRecord(lines[2]).Attributes); !reflect.DeepEqual(got, wantCompletion) {
+		t.Fatalf("completion attribute schema drift:\n got %v\nwant %v", got, wantCompletion)
+	}
+	for _, rl := range lines {
+		if scope := rl.ScopeLogs[0].Scope; scope.Name != otelScopeName || scope.Version != otelAttributeSchemaVersion {
+			t.Fatalf("scope: %+v", scope)
+		}
+	}
+}
+
+func TestOTelWriter_RowValuesAndBody(t *testing.T) {
+	lines := otelSchemaFixture(t)
+	row := otelRecord(lines[0])
+	if v, _ := otelAttr(row.Attributes, "disabled"); v.BoolValue == nil || *v.BoolValue {
+		t.Fatalf("disabled should be boolValue false: %+v", v)
+	}
+	if otelString(t, row.Attributes, "spec") != `{"priority":1000}` {
+		t.Fatalf("nested column must be a JSON string: %+v", row.Attributes)
+	}
+	if _, ok := otelAttr(otelRecord(lines[1]).Attributes, "spec"); ok {
+		t.Fatalf("null column must be dropped from attributes")
+	}
+	if row.Body.StringValue == nil || *row.Body.StringValue != `{"disabled":false,"name":"fw-a","region":"us-east1","spec":{"priority":1000}}` {
+		t.Fatalf("body must be the full row JSON: %+v", row.Body)
+	}
+	if otelString(t, row.Attributes, "stackql.provider") != "google" ||
+		otelString(t, row.Attributes, "stackql.service") != "compute" ||
+		otelString(t, row.Attributes, "stackql.resource") != "firewalls" ||
+		otelString(t, row.Attributes, "stackql.query") != otelTestQuery ||
+		len(otelString(t, row.Attributes, "stackql.query.hash")) != 64 {
+		t.Fatalf("statement context: %+v", row.Attributes)
+	}
+}
+
+func TestOTelWriter_CompletionRecord(t *testing.T) {
+	completion := otelRecord(otelSchemaFixture(t)[2])
+	if v, _ := otelAttr(completion.Attributes, "stackql.rows_returned"); v.IntValue == nil || *v.IntValue != "2" {
+		t.Fatalf("rows_returned: %+v", v)
+	}
+	if v, _ := otelAttr(completion.Attributes, "stackql.snapshot.complete"); v.BoolValue == nil || !*v.BoolValue {
+		t.Fatalf("snapshot.complete: %+v", v)
+	}
+	if completion.Body.StringValue == nil || *completion.Body.StringValue != "snapshot complete google.compute.firewalls" {
+		t.Fatalf("completion body: %+v", completion.Body)
+	}
+}
+
+// Every record of a statement shares the snapshot instant, id, trace and span.
+func TestOTelWriter_StatementIdentity(t *testing.T) {
+	lines := otelSchemaFixture(t)
+	first := otelRecord(lines[0])
+	snapshotID := otelString(t, first.Attributes, "stackql.snapshot.id")
+	for i, rl := range lines {
+		rec := otelRecord(rl)
+		if rec.TimeUnixNano != "1788739200000000000" || rec.TraceID != otelInvocationTraceID() ||
+			rec.SpanID != first.SpanID || otelString(t, rec.Attributes, "stackql.snapshot.id") != snapshotID {
+			t.Fatalf("record %d does not share the statement identity: %+v", i, rec)
+		}
+	}
+}
+
+func TestOTelWriter_ResourceAttributes(t *testing.T) {
+	resource := otelSchemaFixture(t)[0].Resource.Attributes
+	if otelString(t, resource, "service.name") != "stackql" ||
+		otelString(t, resource, "cloud.provider") != "gcp" ||
+		otelString(t, resource, "cloud.account.id") != "p1" ||
+		otelString(t, resource, "cloud.availability_zone") != "z1" {
+		t.Fatalf("resource attributes: %+v", resource)
+	}
+}
+
+func TestOTelWriter_ZeroRowsEmitsOnlyCompletion(t *testing.T) {
+	var out bytes.Buffer
+	w := NewOTelWriter(&out, &bytes.Buffer{}, otelTestQuery, time.Time{})
+	if err := w.Write(otelTestStream()); err != nil {
+		t.Fatalf("Write() error = %v", err)
+	}
+	lines := otelLines(t, out.String())
+	if len(lines) != 1 {
+		t.Fatalf("expected exactly one completion record, got %d", len(lines))
+	}
+	if v, _ := otelAttr(otelRecord(lines[0]).Attributes, "stackql.rows_returned"); v.IntValue == nil || *v.IntValue != "0" {
+		t.Fatalf("rows_returned: %+v", v)
+	}
+}
+
+func TestOTelWriter_FingerprintTracksRowContent(t *testing.T) {
+	fingerprint := func(disabled bool) string {
+		var out bytes.Buffer
+		w := NewOTelWriter(&out, &bytes.Buffer{}, otelTestQuery, time.Time{})
+		if err := w.Write(otelTestStream([]interface{}{"fw-a", "us-east1", disabled, nil})); err != nil {
+			t.Fatalf("Write() error = %v", err)
+		}
+		return otelString(t, otelRecord(otelLines(t, out.String())[0]).Attributes, "stackql.row.fingerprint")
+	}
+	if a, b := fingerprint(false), fingerprint(false); a != b || !strings.HasPrefix(a, "sha256:") {
+		t.Fatalf("identical rows must fingerprint identically: %s vs %s", a, b)
+	}
+	if fingerprint(false) == fingerprint(true) {
+		t.Fatalf("a changed column value must change the fingerprint")
+	}
+}
+
+func TestOTelWriter_WriteError(t *testing.T) {
+	var out, errOut bytes.Buffer
+	w := NewOTelWriter(&out, &errOut, "select 1", time.Time{})
+	if err := w.WriteError(errors.New("boom"), "record"); err != nil {
+		t.Fatalf("WriteError() error = %v", err)
+	}
+	rec := otelRecord(otelLines(t, out.String())[0])
+	if rec.SeverityNumber != sink.OTelSeverityError || rec.SeverityText != "ERROR" ||
+		otelString(t, rec.Attributes, "error.message") != "boom" ||
+		otelString(t, rec.Attributes, "error.type") != "*errors.errorString" {
+		t.Fatalf("error record: %+v", rec)
+	}
+	if err := w.WriteError(errors.New("boom"), stderrPressentationStr); err != nil || strings.TrimSpace(errOut.String()) != "boom" {
+		t.Fatalf("stderr presentation: err=%v out=%q", err, errOut.String())
+	}
+}
+
+func TestGetOutputWriter_OTel(t *testing.T) {
+	ctx := internaldto.OutputContext{RuntimeContext: dto.RuntimeCtx{OutputFormat: "otel"}, Query: "select 1"}
+	w, err := GetOutputWriter(&bytes.Buffer{}, &bytes.Buffer{}, ctx)
+	if err != nil {
+		t.Fatalf("GetOutputWriter() error = %v", err)
+	}
+	if _, ok := w.(*OTelWriter); !ok {
+		t.Fatalf("expected *OTelWriter, got %T", w)
+	}
+}
diff --git a/internal/stackql/output/output.go b/internal/stackql/output/output.go
index 9af75352..7f8c47d1 100644
--- a/internal/stackql/output/output.go
+++ b/internal/stackql/output/output.go
@@ -63,6 +63,8 @@ func GetOutputWriter(
 	case "jsonl", "ndjson":
 		jsonlWriter := NewJSONLWriter(writer, errWriter)
 		return jsonlWriter, nil
+	case otelFormatStr:
+		return NewOTelWriter(writer, errWriter, outputCtx.Query, outputCtx.StartTime), nil
 	case constants.TableStr:
 		tablewriter := TableWriter{
 			AbstractTabularWriter{
diff --git a/internal/stackql/primitivebuilder/shortcuts.go b/internal/stackql/primitivebuilder/shortcuts.go
index c26d325f..3d472cc9 100644
--- a/internal/stackql/primitivebuilder/shortcuts.go
+++ b/internal/stackql/primitivebuilder/shortcuts.go
@@ -11,6 +11,7 @@ import (
 	"github.com/stackql/any-sdk/public/formulation"
 	"github.com/stackql/stackql-parser/go/vt/sqlparser"
 	"github.com/stackql/stackql/internal/stackql/buildinfo"
+	"github.com/stackql/stackql/internal/stackql/contributors"
 	"github.com/stackql/stackql/internal/stackql/handler"
 	"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"
 	"github.com/stackql/stackql/internal/stackql/metadatavisitors"
@@ -251,6 +252,8 @@ func NewShowInstructionExecutor(
 		keys = convertProviderServicesToMap(services, extended)
 	case "VERSION":
 		columnOrder, keys = buildVersionShowOutput(extended)
+	case "CONTRIBUTORS":
+		columnOrder, keys = buildContributorsShowOutput(extended)
 	}
 	return util.PrepareResultSet(internaldto.NewPrepareResultSetDTO(nil, keys, columnOrder, nil, err, nil,
 		handlerCtx.GetTypingConfig()))
@@ -340,6 +343,23 @@ func buildVersionShowOutput(extended bool) ([]string, map[string]map[string]inte
 		}
 }
 
+// buildContributorsShowOutput renders SHOW [EXTENDED] CONTRIBUTORS; keys are
+// zero-padded so the sorted result set keeps leaderboard order.
+func buildContributorsShowOutput(extended bool) ([]string, map[string]map[string]interface{}) {
+	keys := make(map[string]map[string]interface{})
+	for i, c := range contributors.List() {
+		row := map[string]interface{}{"contributor": c.GetLogin()}
+		if extended {
+			row["contributions"] = c.GetContributions()
+		}
+		keys[fmt.Sprintf("%06d", i)] = row
+	}
+	if extended {
+		return []string{"contributor", "contributions"}, keys
+	}
+	return []string{"contributor"}, keys
+}
+
 //nolint:errcheck // future proofing
 func filterResources(
 	resources map[string]formulation.Resource,
diff --git a/internal/stackql/primitivegenerator/primitive_generator.go b/internal/stackql/primitivegenerator/primitive_generator.go
index 8696caea..f1a070b0 100644
--- a/internal/stackql/primitivegenerator/primitive_generator.go
+++ b/internal/stackql/primitivegenerator/primitive_generator.go
@@ -272,7 +272,7 @@ func (pb *standardPrimitiveGenerator) inferProviderForShow(
 		pb.PrimitiveComposer.SetProvider(prov)
 	case "PROVIDERS":
 		// no provider, might create some dummy object dunno
-	case "VERSION":
+	case "VERSION", "CONTRIBUTORS":
 		// no provider needed
 	case "RESOURCES":
 		prov, err := handlerCtx.GetProvider(node.OnTable.Qualifier.GetRawVal())
diff --git a/internal/stackql/primitivegenerator/statement_analyzer.go b/internal/stackql/primitivegenerator/statement_analyzer.go
index e276b0bf..d51d0f5a 100644
--- a/internal/stackql/primitivegenerator/statement_analyzer.go
+++ b/internal/stackql/primitivegenerator/statement_analyzer.go
@@ -1668,7 +1668,7 @@ func (pb *standardPrimitiveGenerator) analyzeShow(
 		return nil
 	case "PROVIDERS":
 		// TODO
-	case "VERSION":
+	case "VERSION", "CONTRIBUTORS":
 		// no further analysis required
 	case "RESOURCES":
 		prov, err := handlerCtx.GetProvider(node.OnTable.Qualifier.GetRawVal())
diff --git a/internal/stackql/responsehandler/responsehandler.go b/internal/stackql/responsehandler/responsehandler.go
index dcb22c9d..30e06483 100644
--- a/internal/stackql/responsehandler/responsehandler.go
+++ b/internal/stackql/responsehandler/responsehandler.go
@@ -37,6 +37,8 @@ func HandleResponse(handlerCtx handler.HandlerContext, response internaldto.Exec
 			internaldto.OutputContext{
 				RuntimeContext: handlerCtx.GetRuntimeContext(),
 				Result:         sqlResult,
+				Query:          handlerCtx.GetRawQuery(),
+				StartTime:      handlerCtx.GetQueryStartTime(),
 			},
 		)
 		if outputWriter == nil || err != nil {
@@ -51,6 +53,8 @@ func HandleResponse(handlerCtx handler.HandlerContext, response internaldto.Exec
 			internaldto.OutputContext{
 				RuntimeContext: handlerCtx.GetRuntimeContext(),
 				Result:         sqlResult,
+				Query:          handlerCtx.GetRawQuery(),
+				StartTime:      handlerCtx.GetQueryStartTime(),
 			},
 		)
 		if outputWriter == nil || err != nil {
diff --git a/pkg/sink/otel.go b/pkg/sink/otel.go
index 7608feb0..cdae697a 100644
--- a/pkg/sink/otel.go
+++ b/pkg/sink/otel.go
@@ -1,6 +1,7 @@
 package sink
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"encoding/hex"
@@ -222,25 +223,37 @@ func genericOTelLogRecord(payload any, now time.Time) (OTelLogRecord, error) {
 		return OTelLogRecord{}, fmt.Errorf("otel sink: marshal payload: %w", err)
 	}
 	record := OTelLogRecord{Time: now, Severity: OTelSeverityInfo, Body: string(raw)}
+	// Not a JSON object (scalar or array): the body alone carries it.
+	record.Attributes, _ = OTelAttributesFromJSON(raw)
+	return record, nil
+}
+
+// OTelAttributesFromJSON maps a JSON object's top-level fields to typed
+// attributes in key order; nested values are carried as JSON strings and
+// nulls as empty strings (dropped on encode). ok is false when raw is not
+// an object.
+func OTelAttributesFromJSON(raw []byte) ([]OTelAttribute, bool) {
 	var fields map[string]json.RawMessage
-	if unmarshalErr := json.Unmarshal(raw, &fields); unmarshalErr != nil {
-		// Not a JSON object (scalar or array): the body alone carries it.
-		return record, nil //nolint:nilerr // intentional: non-object payloads are still valid records
+	if err := json.Unmarshal(raw, &fields); err != nil {
+		return nil, false
 	}
 	keys := make([]string, 0, len(fields))
 	for k := range fields {
 		keys = append(keys, k)
 	}
 	sort.Strings(keys)
+	attrs := make([]OTelAttribute, 0, len(keys))
 	for _, k := range keys {
-		record.Attributes = append(record.Attributes, otelAttributeFromJSON(k, fields[k]))
+		attrs = append(attrs, otelAttributeFromJSON(k, fields[k]))
 	}
-	return record, nil
+	return attrs, true
 }
 
 func otelAttributeFromJSON(key string, raw json.RawMessage) OTelAttribute {
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber()
 	var v any
-	if err := json.Unmarshal(raw, &v); err != nil {
+	if err := dec.Decode(&v); err != nil {
 		return OTelString(key, string(raw))
 	}
 	switch t := v.(type) {
@@ -248,11 +261,18 @@ func otelAttributeFromJSON(key string, raw json.RawMessage) OTelAttribute {
 		return OTelString(key, t)
 	case bool:
 		return OTelBool(key, t)
-	case float64:
-		if t == float64(int64(t)) {
-			return OTelInt(key, int64(t))
+	case json.Number:
+		if i, intErr := t.Int64(); intErr == nil {
+			return OTelInt(key, i)
+		}
+		f, floatErr := t.Float64()
+		if floatErr != nil {
+			return OTelString(key, t.String())
+		}
+		if f == float64(int64(f)) {
+			return OTelInt(key, int64(f))
 		}
-		return OTelDouble(key, t)
+		return OTelDouble(key, f)
 	case nil:
 		return OTelString(key, "")
 	default:
diff --git a/test/python/stackql_test_tooling/StackQLInterfaces.py b/test/python/stackql_test_tooling/StackQLInterfaces.py
index 2dd8b0b2..f6466976 100644
--- a/test/python/stackql_test_tooling/StackQLInterfaces.py
+++ b/test/python/stackql_test_tooling/StackQLInterfaces.py
@@ -1080,6 +1080,83 @@ def _parse_jsonl(self, payload :str, label :str) -> list:
       rows.append(json.dumps(parsed, sort_keys=True))
     return sorted(rows)
 
+  @keyword
+  def should_stackql_exec_otel_snapshot(
+    self,
+    stackql_exe :str,
+    okta_secret_str :str,
+    github_secret_str :str,
+    k8s_secret_str :str,
+    registry_cfg :RegistryCfg,
+    auth_cfg_str :str,
+    sql_backend_cfg_str :str,
+    query :str,
+    expected :str,
+    *args,
+    **cfg
+  ) -> typing.List[str]:
+    """
+    Run with `--output otel` and check the snapshot invariants (issue #738):
+    one LogsData per row plus one completion record, every record sharing the
+    snapshot instant, snapshot id and span. `expected` is a JSON object with
+    the known attribute values: `context` is verified on every record, `rows`
+    (in order) on the row records and `completion` on the completion record;
+    dynamic values (ids, timestamps, fingerprints) are simply left unlisted.
+    Returns the row fingerprints so a repeat run can be compared.
+    """
+    expectation = json.loads(expected)
+    result = self._run_stackql_exec_command(
+      stackql_exe, okta_secret_str, github_secret_str, k8s_secret_str,
+      registry_cfg, auth_cfg_str, sql_backend_cfg_str, query, '-o=otel', *args, **cfg
+    )
+    records = []
+    for line in result.stdout.splitlines():
+      if not line.strip():
+        continue
+      scope_logs = json.loads(line)['resourceLogs'][0]['scopeLogs'][0]
+      if scope_logs['scope']['name'] != 'github.com/stackql/stackql/internal/stackql/output':
+        raise Exception(f'unexpected scope: {scope_logs["scope"]}')
+      for rec in scope_logs['logRecords']:
+        rec['attrs'] = {a['key']: self._otel_value(a['value']) for a in rec['attributes']}
+        records.append(rec)
+    rows = [r for r in records if 'stackql.row.index' in r['attrs']]
+    completions = [r for r in records if r['attrs'].get('stackql.snapshot.complete') is True]
+    expected_rows = expectation.get('rows', [])
+    if len(rows) != len(expected_rows) or len(completions) != 1 or len(records) != len(rows) + 1:
+      raise Exception(
+        f'expected {len(expected_rows)} row records and one completion record, '
+        f'got {len(rows)} rows and {len(completions)} completions in {len(records)} records; '
+        f'stderr: {(result.stderr or "").strip()[-2000:]}'
+      )
+    for i, (rec, want) in enumerate(zip(rows + completions, expected_rows + [expectation.get('completion', {})])):
+      label = f'row {i}' if i < len(rows) else 'completion'
+      for key, value in {**expectation.get('context', {}), **want}.items():
+        if key not in rec['attrs']:
+          raise Exception(f'{label}: attribute {key!r} missing; have {sorted(rec["attrs"])}')
+        if rec['attrs'][key] != value:
+          raise Exception(f'{label}: attribute {key!r} = {rec["attrs"][key]!r}, want {value!r}')
+    for key in ('timeUnixNano', 'spanId', 'traceId'):
+      if len({r[key] for r in records}) != 1:
+        raise Exception(f'{key} differs across the records of one statement')
+    if len({r['attrs']['stackql.snapshot.id'] for r in records}) != 1:
+      raise Exception('stackql.snapshot.id differs across the records of one statement')
+    if [r['attrs']['stackql.row.index'] for r in rows] != list(range(len(rows))):
+      raise Exception(f'row indices are not sequential: {[r["attrs"]["stackql.row.index"] for r in rows]}')
+    if completions[0]['attrs']['stackql.rows_returned'] != len(rows):
+      raise Exception(f'completion rows_returned {completions[0]["attrs"]["stackql.rows_returned"]} != {len(rows)}')
+    fingerprints = [r['attrs']['stackql.row.fingerprint'] for r in rows]
+    if any(not fp.startswith('sha256:') or len(fp) != 71 for fp in fingerprints):
+      raise Exception(f'malformed fingerprints: {fingerprints}')
+    return fingerprints
+
+  @staticmethod
+  def _otel_value(any_value :dict):
+    """Decode an OTLP/JSON AnyValue to the natural python type."""
+    kind, raw = next(iter(any_value.items()))
+    if kind == 'intValue':
+      return int(raw)
+    return raw
+
   @keyword
   def should_stackql_exec_inline_equal_both_streams(
     self, 
diff --git a/test/robot/functional/mcp.robot b/test/robot/functional/mcp.robot
index 815fb72f..2cdeb270 100644
--- a/test/robot/functional/mcp.robot
+++ b/test/robot/functional/mcp.robot
@@ -328,6 +328,27 @@ PG Server Show Version Extended
     Should Match Regexp    ${psql_client_result.stdout}       \\d+\\.\\d+\\.\\d+
     Should Be Equal As Integers    ${psql_client_result.rc}    0
 
+PG Server Show Contributors
+    [Documentation]    Issue #320: the embedded leaderboard is served over the
+    ...                wire protocol from the same implementation.
+    Pass Execution If    "%{IS_SKIP_MCP_TEST=false}" == "true"    Some platforms do not have the MCP client available
+    Sleep         5s
+    ${posixInput} =     Catenate
+    ...    "${PSQL_EXE}"    -d     postgres://stackql:stackql@127.0.0.1:5665   -c
+    ...    "SHOW EXTENDED CONTRIBUTORS;"
+    ${windowsInput} =     Catenate
+    ...    &    ${posixInput}
+    ${input} =    Set Variable If    "${IS_WINDOWS}" == "1"    ${windowsInput}    ${posixInput}
+    ${shellExe} =    Set Variable If    "${IS_WINDOWS}" == "1"    powershell    sh
+    ${psql_client_result}=    Run Process
+    ...                  ${shellExe}     \-c    ${input}
+    ...                  stdout=${CURDIR}${/}tmp${/}PG-Server-Show-Contributors-psql.txt
+    ...                  stderr=${CURDIR}${/}tmp${/}PG-Server-Show-Contributors-psql-stderr.txt
+    Should Contain       ${psql_client_result.stdout}       contributor
+    Should Contain       ${psql_client_result.stdout}       contributions
+    Should Match Regexp    ${psql_client_result.stdout}       (?s)general-kroll-4-life.*jeffreyaven
+    Should Be Equal As Integers    ${psql_client_result.rc}    0
+
 MCP HTTP Server Query Tool
     Pass Execution If    "%{IS_SKIP_MCP_TEST=false}" == "true"    Some platforms do not have the MCP client available
     Sleep         5s
diff --git a/test/robot/functional/stackql_from_cmd_line.robot b/test/robot/functional/stackql_from_cmd_line.robot
index 0ded8ec8..b69eeb05 100644
--- a/test/robot/functional/stackql_from_cmd_line.robot
+++ b/test/robot/functional/stackql_from_cmd_line.robot
@@ -810,6 +810,26 @@ Show Version Extended
     ${captured}=    Get File    ${CURDIR}/tmp/Show-Version-Extended.tmp
     Should Match Regexp    ${captured}    \\d+\\.\\d+\\.\\d+
 
+Show Contributors
+    [Documentation]    Issue #320: SHOW CONTRIBUTORS renders the embedded
+    ...                leaderboard, contributions descending, founder on top.
+    Should StackQL Exec Contain
+    ...    SHOW CONTRIBUTORS;
+    ...    contributor
+    ...    stdout=${CURDIR}/tmp/Show-Contributors.tmp
+    ${captured}=    Get File    ${CURDIR}/tmp/Show-Contributors.tmp
+    Should Match Regexp    ${captured}    (?s)general-kroll-4-life.*jeffreyaven
+    Should Not Contain    ${captured}    contributions
+
+Show Contributors Extended
+    [Documentation]    SHOW EXTENDED CONTRIBUTORS adds the contributions count.
+    Should StackQL Exec Contain
+    ...    SHOW EXTENDED CONTRIBUTORS;
+    ...    contributions
+    ...    stdout=${CURDIR}/tmp/Show-Contributors-Extended.tmp
+    ${captured}=    Get File    ${CURDIR}/tmp/Show-Contributors-Extended.tmp
+    Should Match Regexp    ${captured}    (?s)general-kroll-4-life.*jeffreyaven
+
 *** Keywords ***
 Should StackQL Exec Equal
     [Arguments]    ${_EXEC_CMD_STR}    ${_EXEC_CMD_EXPECTED_OUTPUT}    @{varargs}    &{kwargs}
diff --git a/test/robot/functional/stackql_mocked_from_cmd_line.robot b/test/robot/functional/stackql_mocked_from_cmd_line.robot
index 470384eb..5c7182d2 100644
--- a/test/robot/functional/stackql_mocked_from_cmd_line.robot
+++ b/test/robot/functional/stackql_mocked_from_cmd_line.robot
@@ -10881,6 +10881,36 @@ Preview Rows Are Emitted Throughout The Run
     ...    stdout=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run.tmp
     ...    stderr=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-stderr.tmp
 
+Preview Rows Are Emitted Throughout The Run As OTel Records
+    [Documentation]    Issue #738: the otel writer streams like jsonl, one
+    ...                LogsData per row as the upstream produces it, then the
+    ...                completion record; 79 rows plus completion is 80 lines.
+    [Setup]    Write Gcp Service Account    ${OMNISDK_MOCK_GCP_SA_HOST}
+    [Teardown]    Remove Preview Mock Environment
+    ${mock} =    Set Variable    {"scheme":"http","host":"${LOCAL_HOST_ALIAS}","port":"${MOCKSERVER_PORT_OMNISDK}"}
+    ${preview} =    Catenate    SEPARATOR=
+    ...    {"batchSize":10,"flushInterval":"50ms","endpoint":{"aws.s3":${mock}}}
+    Set Environment Variable    AWS_ACCESS_KEY_ID    AK
+    Set Environment Variable    AWS_SECRET_ACCESS_KEY    SK
+    ${query} =    Catenate    SEPARATOR=${SPACE}
+    ...    select * from stackql_preview.audit.aws_s3_buckets
+    ...    where region = 'us-east-1' and method = 'list';
+    Should StackQL Exec Stream Incrementally
+    ...    ${STACKQL_EXE}
+    ...    ${OKTA_SECRET_STR}
+    ...    ${GITHUB_SECRET_STR}
+    ...    ${K8S_SECRET_STR}
+    ...    ${REGISTRY_NO_VERIFY_CFG_STR}
+    ...    ${AUTH_CFG_STR}
+    ...    ${SQL_BACKEND_CFG_STR_CANONICAL}
+    ...    ${query}
+    ...    80
+    ...    0.2
+    ...    \-o\=otel
+    ...    --preview\=${preview}
+    ...    stdout=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-As-OTel-Records.tmp
+    ...    stderr=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-As-OTel-Records-stderr.tmp
+
 Preview Order By Is Refused Rather Than Ignored
     [Documentation]    A streamed relation never reaches the SQL backend, so
     ...                ORDER BY cannot be applied. It is refused outright rather
@@ -11017,3 +11047,76 @@ Unstable Google Kms Key Rings Jsonl Row Set Matches Expectation
     ...    --preview\=${preview}
     ...    stdout=${CURDIR}${/}tmp${/}Unstable-Google-Kms-Key-Rings.tmp
     ...    stderr=${CURDIR}${/}tmp${/}Unstable-Google-Kms-Key-Rings-stderr.tmp
+
+OTel Output Emits One Record Per Row Plus Completion
+    [Documentation]    Issue #738: --output otel writes one OTLP/JSON LogsData per
+    ...                row and a completion record with the row values as typed
+    ...                attributes. Every record of the statement shares the
+    ...                snapshot instant, snapshot id and span, and an identical
+    ...                re-run yields identical row fingerprints.
+    ${query} =    Catenate    SEPARATOR=${SPACE}
+    ...    select login, contributions from github.repos.contributors
+    ...    where owner = 'dummyorg' and repo = 'dummyapp.io';
+    ${expected} =    Catenate    SEPARATOR=
+    ...    {"context":{"stackql.provider":"github","stackql.service":"repos","stackql.resource":"contributors",
+    ...    "stackql.query":"${query}"},
+    ...    "rows":[
+    ...    {"stackql.row.index":0,"login":"octo-lead","contributions":"500"},
+    ...    {"stackql.row.index":1,"login":"octo-dev-one","contributions":"120"},
+    ...    {"stackql.row.index":2,"login":"octo-dev-two","contributions":"120"},
+    ...    {"stackql.row.index":3,"login":"octo-maintainer","contributions":"80"},
+    ...    {"stackql.row.index":4,"login":"octo-helper-one","contributions":"45"},
+    ...    {"stackql.row.index":5,"login":"octo-helper-two","contributions":"45"},
+    ...    {"stackql.row.index":6,"login":"octo-casual","contributions":"12"},
+    ...    {"stackql.row.index":7,"login":"octo-drive-by","contributions":"3"}],
+    ...    "completion":{"stackql.snapshot.complete":true,"stackql.rows_returned":8}}
+    ${first} =    Should StackQL Exec OTel Snapshot
+    ...    ${STACKQL_EXE}
+    ...    ${OKTA_SECRET_STR}
+    ...    ${GITHUB_SECRET_STR}
+    ...    ${K8S_SECRET_STR}
+    ...    ${REGISTRY_NO_VERIFY_CFG_STR}
+    ...    ${AUTH_CFG_STR}
+    ...    ${SQL_BACKEND_CFG_STR_CANONICAL}
+    ...    ${query}
+    ...    ${expected}
+    ...    stdout=${CURDIR}${/}tmp${/}OTel-Output-Rows-First.tmp
+    ...    stderr=${CURDIR}${/}tmp${/}OTel-Output-Rows-First-stderr.tmp
+    ${second} =    Should StackQL Exec OTel Snapshot
+    ...    ${STACKQL_EXE}
+    ...    ${OKTA_SECRET_STR}
+    ...    ${GITHUB_SECRET_STR}
+    ...    ${K8S_SECRET_STR}
+    ...    ${REGISTRY_NO_VERIFY_CFG_STR}
+    ...    ${AUTH_CFG_STR}
+    ...    ${SQL_BACKEND_CFG_STR_CANONICAL}
+    ...    ${query}
+    ...    ${expected}
+    ...    stdout=${CURDIR}${/}tmp${/}OTel-Output-Rows-Second.tmp
+    ...    stderr=${CURDIR}${/}tmp${/}OTel-Output-Rows-Second-stderr.tmp
+    Should Be Equal    ${first}    ${second}
+
+OTel Output Zero Row Statement Emits Only Completion
+    [Documentation]    Issue #738: a statement returning no rows still emits its
+    ...                completion record, so an absent resource is visible as
+    ...                drift rather than as silence.
+    ${query} =    Catenate    SEPARATOR=${SPACE}
+    ...    select login from github.repos.contributors
+    ...    where owner = 'dummyorg' and repo = 'dummyapp.io' and login = 'nobody';
+    ${expected} =    Catenate    SEPARATOR=
+    ...    {"context":{"stackql.provider":"github","stackql.service":"repos","stackql.resource":"contributors",
+    ...    "stackql.query":"${query}"},
+    ...    "rows":[],
+    ...    "completion":{"stackql.snapshot.complete":true,"stackql.rows_returned":0}}
+    Should StackQL Exec OTel Snapshot
+    ...    ${STACKQL_EXE}
+    ...    ${OKTA_SECRET_STR}
+    ...    ${GITHUB_SECRET_STR}
+    ...    ${K8S_SECRET_STR}
+    ...    ${REGISTRY_NO_VERIFY_CFG_STR}
+    ...    ${AUTH_CFG_STR}
+    ...    ${SQL_BACKEND_CFG_STR_CANONICAL}
+    ...    ${query}
+    ...    ${expected}
+    ...    stdout=${CURDIR}${/}tmp${/}OTel-Output-Zero-Rows.tmp
+    ...    stderr=${CURDIR}${/}tmp${/}OTel-Output-Zero-Rows-stderr.tmp