First successfull run via Arrow

Author: vitalyisaev2

app/server/config/config.debug.yaml

@@ -81,7 +81,7 @@ datasources:
   ydb:
     <<: *data_source_default_var
     use_underlay_network_for_dedicated_databases: false
-    mode: MODE_QUERY_SERVICE_NATIVE
+    mode: MODE_QUERY_SERVICE_NATIVE_ARROW
     pushdown:
       enable_timestamp_pushdown: true
     splitting:

app/server/config/config.go

@@ -544,10 +544,10 @@ func validateYdbConfig(c *config.TYdbConfig) error {
 	}
 
 	switch c.Mode {
-	case config.TYdbConfig_MODE_QUERY_SERVICE_NATIVE:
-		if c.ResourcePool == "" {
-			return fmt.Errorf("you must set `resource_pool` if `mode` is `query_service_native`")
-		}
+	case config.TYdbConfig_MODE_QUERY_SERVICE_NATIVE, config.TYdbConfig_MODE_QUERY_SERVICE_NATIVE_ARROW:
+		// if c.ResourcePool == "" {
+		// 	return fmt.Errorf("you must set `resource_pool` if `mode` is `query_service_native` or `query_service_native_arrow`")
+		// }
 	case config.TYdbConfig_MODE_TABLE_SERVICE_STDLIB_SCAN_QUERIES:
 		if c.ResourcePool != "" {
 			return fmt.Errorf("you must not set `resource_pool` if `mode` is `table_service_stdlib_scan_queries`")

app/server/datasource/rdbms/ydb/connection_manager.go

@@ -58,7 +58,8 @@ func (c *connectionManager) Make(
 	} else if dsi.Credentials.GetBasic() != nil {
 		logger.Debug("connector will use base auth credentials for authorization")
 
-		cred = ydb_sdk.WithStaticCredentials(dsi.Credentials.GetBasic().Username, dsi.Credentials.GetBasic().Password)
+		//cred = ydb_sdk.WithStaticCredentials(dsi.Credentials.GetBasic().Username, dsi.Credentials.GetBasic().Password)
+		cred = ydb_sdk.WithAnonymousCredentials()
 	} else {
 		logger.Warn("connector will not use any credentials for authorization")
 

app/server/paging/interface.go

@@ -43,6 +43,7 @@ type ColumnarBufferFactory[T Acceptor] interface {
 // 4. flag marking this stream as completed
 type ReadResult[T Acceptor] struct {
 	ColumnarBuffer    ColumnarBuffer[T]
+	Data              []byte // serialized Arrow Record
 	Stats             *api_service_protos.TReadSplitsResponse_TStats
 	Error             error
 	IsTerminalMessage bool

app/server/paging/sink.go

@@ -215,18 +215,13 @@ func (s *sinkImpl[T]) respondWithArrowRecord(
 		return
 	}
 
-	// Create a columnar buffer for the response
-	// We'll create a simple implementation that just returns the response
-	cb := &columnarBufferArrowIPCStreamingDefault[T]{
-		arrowAllocator: nil,
-		builders:       nil,
-		schema:         record.Schema(),
-		logger:         s.logger,
-	}
+	// Get the serialized data from the buffer
+	serializedData := buf.Bytes()
 
-	// Create a result with the response
+	// Create a result with the serialized data
 	result := &ReadResult[T]{
-		ColumnarBuffer:    cb,
+		ColumnarBuffer:    nil,
+		Data:              serializedData,
 		Stats:             stats,
 		Error:             err,
 		IsTerminalMessage: isTerminalMessage,

app/server/streaming/read_splits_streamer.go

@@ -60,13 +60,30 @@ func (s *ReadSplitsStreamer[T]) writeDataToStream() error {
 }
 
 func (s *ReadSplitsStreamer[T]) sendResultToStream(result *paging.ReadResult[T]) error {
-	// buffer must be explicitly marked as unused,
-	// otherwise memory will leak
-	defer result.ColumnarBuffer.Release()
+	var resp *api_service_protos.TReadSplitsResponse
 
-	resp, err := result.ColumnarBuffer.ToResponse()
-	if err != nil {
-		return fmt.Errorf("buffer to response: %w", err)
+	var err error
+
+	if result.Data != nil {
+		fmt.Println(" >>> HERE <<<")
+		// Handle the case where we have serialized Arrow data
+		resp = &api_service_protos.TReadSplitsResponse{
+			Payload: &api_service_protos.TReadSplitsResponse_ArrowIpcStreaming{
+				ArrowIpcStreaming: result.Data,
+			},
+		}
+	} else if result.ColumnarBuffer != nil {
+		// Handle the case where we have a columnar buffer
+		// buffer must be explicitly marked as unused,
+		// otherwise memory will leak
+		defer result.ColumnarBuffer.Release()
+
+		resp, err = result.ColumnarBuffer.ToResponse()
+		if err != nil {
+			return fmt.Errorf("buffer to response: %w", err)
+		}
+	} else {
+		return fmt.Errorf("result contains neither Data nor ColumnarBuffer")
 	}
 
 	resp.Stats = result.Stats

scripts/debug/kqprun/scheme.YQ-4416.txt

@@ -0,0 +1,10 @@
+CREATE OBJECT ydb_local_password (TYPE SECRET) WITH (value = password);
+
+CREATE EXTERNAL DATA SOURCE external_datasource WITH (
+    SOURCE_TYPE="Ydb",
+    LOCATION="localhost:2136",
+    AUTH_METHOD="BASIC",
+    LOGIN="admin",
+    DATABASE_NAME="/Root",
+    PASSWORD_SECRET_NAME="ydb_local_password"
+);

scripts/debug/kqprun/schema.redis.local.txt renamed to scripts/debug/kqprun/scheme.redis.local.txt

@@ -0,0 +1,4 @@
+PRAGMA generic.UsePredicatePushdown="true";
+
+SELECT * FROM external_datasource.olap_lineitem_s10 LIMIT 10;
+