feat(go/adbc/driver/flightsql): add logging for Flight SQL

c/driver_manager/adbc_driver_manager.cc

@@ -16,11 +16,19 @@
 // under the License.
 
 #if defined(_WIN32)
-#include <windows.h>  // Must come first
-
+// These version macros gate which Win32 APIs the SDK headers declare. They MUST
+// be set before <windows.h> is included -- once windows.h pulls in winnt.h, the
+// internal API-availability macros are fixed and later #defines have no effect.
+// In particular, SHGetKnownFolderPath (used below) requires _WIN32_WINNT >= 0x0600
+// (Vista). Without this, builds with toolchains that default _WIN32_WINNT below
+// Vista (e.g. TDM-GCC 10.x) fail with "SHGetKnownFolderPath was not declared".
+#ifndef _WIN32_WINNT
+#define _WIN32_WINNT 0x0A00  // Windows 10
+#endif
 #ifndef NTDDI_VERSION
 #define NTDDI_VERSION 0x0A00000C  // For SHGetKnownFolderPath in ShlObj_core.h in ShlObj.h
 #endif
+#include <windows.h>  // Must come first
 
 #include <KnownFolders.h>
 #include <ShlObj.h>

go/adbc/adbc.go

@@ -264,6 +264,28 @@ const (
 	OptionKeyPassword                 = "password"
 	// EXPERIMENTAL. Sets/Gets the trace parent on OpenTelemetry traces
 	OptionKeyTelemetryTraceParent = "adbc.telemetry.trace_parent"
+	// EXPERIMENTAL. Selects the OpenTelemetry traces exporter when the
+	// driver initializes its tracer provider. Accepts the same values as
+	// the OpenTelemetry "OTEL_TRACES_EXPORTER" environment variable (see
+	// the OptionTelemetryExporter constants below: "none", "otlp",
+	// "console", "adbcfile"). When this option is set on a database it
+	// takes precedence over the OTEL_TRACES_EXPORTER environment
+	// variable, which lets operators select an exporter via the ADBC
+	// driver manager / TOML profile mechanism without having to mutate
+	// the host process's environment. When neither this option nor the
+	// environment variable is set, the driver falls back to the
+	// process-global OpenTelemetry tracer provider.
+	OptionKeyTelemetryTracesExporter = "adbc.telemetry.traces_exporter"
+	// EXPERIMENTAL. Selects the on-disk folder used by the "adbcfile"
+	// traces exporter. When the exporter is "adbcfile" and this option
+	// is set, rotated trace files are written to the supplied folder
+	// (which is created if it does not exist) instead of the default
+	// "<user-config-dir>/.adbc/traces" path. The option is ignored for
+	// other exporters; it exists so an operator can route trace files
+	// to a location their support workflow already collects (e.g. a
+	// shared diagnostics folder) via the ADBC driver-manager / TOML
+	// profile mechanism.
+	OptionKeyTelemetryTracesFolderPath = "adbc.telemetry.traces_folder_path"
 )
 
 // EXPERIMENTAL. Traces Telemetry exporter option type

go/adbc/driver/flightsql/flightsql_adbc_server_test.go

@@ -729,7 +729,12 @@ func (ts *ErrorDetailsTests) TestBinaryDetails() {
 
 	ts.Equal(int32(codes.FailedPrecondition), adbcErr.VendorCode)
 
-	ts.Equal(2, len(adbcErr.Details))
+	// Expected detail set:
+	//   - x-header-bin, x-trailer-bin: server-supplied details
+	//   - grpc_code, grpc_message: canonical gRPC code/message added
+	//     by adbcFromFlightStatusWithDetails for every Flight error
+	//   - statement_id, connection_id: driver-attached correlation IDs
+	ts.Equal(6, len(adbcErr.Details))
 
 	headerFound := false
 	trailerFound := false
@@ -745,6 +750,14 @@ func (ts *ErrorDetailsTests) TestBinaryDetails() {
 			ts.NoError(err)
 			ts.Equal([]byte{111, 0, 112}, val)
 			trailerFound = true
+		case "statement_id", "connection_id":
+			val, err := wrapper.Serialize()
+			ts.NoError(err)
+			ts.NotEmpty(val)
+		case "grpc_code", "grpc_message":
+			val, err := wrapper.Serialize()
+			ts.NoError(err)
+			ts.NotEmpty(val)
 		default:
 			ts.Failf("Unexpected detail key: %s", wrapper.Key())
 		}
@@ -766,9 +779,21 @@ func (ts *ErrorDetailsTests) TestGetFlightInfo() {
 
 	ts.Equal(int32(codes.Unknown), adbcErr.VendorCode)
 
-	ts.Equal(1, len(adbcErr.Details))
-
-	wrapper := adbcErr.Details[0]
+	// Expected detail set:
+	//   - grpc-status-details-bin: one server-supplied protobuf detail
+	//   - grpc_code, grpc_message: canonical gRPC code/message added
+	//     by adbcFromFlightStatusWithDetails for every Flight error
+	//   - statement_id, connection_id: driver-attached correlation IDs
+	ts.Equal(5, len(adbcErr.Details))
+
+	var wrapper adbc.ErrorDetail
+	for _, d := range adbcErr.Details {
+		if d.Key() == "grpc-status-details-bin" {
+			wrapper = d
+			break
+		}
+	}
+	ts.NotNil(wrapper, "grpc-status-details-bin detail not found")
 	ts.Equal("grpc-status-details-bin", wrapper.Key())
 
 	raw, err := wrapper.Serialize()
@@ -801,9 +826,21 @@ func (ts *ErrorDetailsTests) TestDoGet() {
 	var adbcErr adbc.Error
 	ts.ErrorAs(err, &adbcErr, "Error was: %#v", err)
 
-	ts.Equal(1, len(adbcErr.Details))
-
-	wrapper := adbcErr.Details[0]
+	// Expected detail set:
+	//   - grpc-status-details-bin: one server-supplied protobuf detail
+	//   - grpc_code, grpc_message: canonical gRPC code/message added
+	//     by adbcFromFlightStatusWithDetails for every Flight error
+	//   - statement_id, connection_id: driver-attached correlation IDs
+	ts.Equal(5, len(adbcErr.Details))
+
+	var wrapper adbc.ErrorDetail
+	for _, d := range adbcErr.Details {
+		if d.Key() == "grpc-status-details-bin" {
+			wrapper = d
+			break
+		}
+	}
+	ts.NotNil(wrapper, "grpc-status-details-bin detail not found")
 	ts.Equal("grpc-status-details-bin", wrapper.Key())
 
 	raw, err := wrapper.Serialize()

go/adbc/driver/flightsql/flightsql_bulk_ingest.go

@@ -20,6 +20,8 @@ package flightsql
 import (
 	"context"
 	"fmt"
+	"log/slog"
+	"time"
 
 	"github.com/apache/arrow-adbc/go/adbc"
 	"github.com/apache/arrow-go/v18/arrow"
@@ -111,6 +113,26 @@ func (s *statement) executeIngest(ctx context.Context) (int64, error) {
 		}
 	}
 
+	startTime := time.Now()
+	catalogStr := ""
+	if s.catalog != nil {
+		catalogStr = *s.catalog
+	}
+	dbSchemaStr := ""
+	if s.dbSchema != nil {
+		dbSchemaStr = *s.dbSchema
+	}
+	startAttrs := []any{
+		slog.String("target_table", s.targetTable),
+		slog.String("mode", s.ingestMode),
+		slog.String("catalog", catalogStr),
+		slog.String("db_schema", dbSchemaStr),
+		slog.Bool("temporary", s.temporary),
+		slog.Bool("streamBind", s.streamBind != nil),
+		slog.Bool("recordBound", s.bound != nil),
+	}
+	s.log.InfoContext(ctx, "FlightSQL ExecuteIngest start", startAttrs...)
+
 	opts := ingestOptions{
 		targetTable: s.targetTable,
 		mode:        s.ingestMode,
@@ -129,6 +151,10 @@ func (s *statement) executeIngest(ctx context.Context) (int64, error) {
 	} else {
 		rdr, err = createRecordReaderFromBatch(s.bound)
 		if err != nil {
+			s.log.WarnContext(ctx, "FlightSQL ExecuteIngest finished with error",
+				slog.Duration("duration", time.Since(startTime)),
+				"err", err,
+			)
 			return -1, err
 		}
 	}
@@ -138,9 +164,21 @@ func (s *statement) executeIngest(ctx context.Context) (int64, error) {
 	callOpts := append([]grpc.CallOption{}, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts)
 
 	nRows, err := s.cnxn.cl.ExecuteIngest(ctx, rdr, ingestOpts, callOpts...)
+	finishAttrs := []any{
+		slog.Duration("duration", time.Since(startTime)),
+		slog.Int64("rowsIngested", nRows),
+	}
+	finishAttrs = append(finishAttrs, correlationHeaderAttrs(header)...)
+	finishAttrs = append(finishAttrs, correlationHeaderAttrs(trailer)...)
 	if err != nil {
-		return -1, adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteIngest")
+		wrapped := withOperationIDs(
+			adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteIngest"),
+			s.id, s.cnxn.id)
+		finishAttrs = append(finishAttrs, "err", wrapped)
+		s.log.WarnContext(ctx, "FlightSQL ExecuteIngest finished with error", finishAttrs...)
+		return -1, wrapped
 	}
+	s.log.InfoContext(ctx, "FlightSQL ExecuteIngest finished", finishAttrs...)
 
 	return nRows, nil
 }