[SPARK-41196] [CONNECT] Homogenize the protobuf version across the Spark connect server to use the same major version.

connector/connect/src/main/buf.gen.yaml

@@ -27,7 +27,7 @@ plugins:
   - remote: buf.build/protocolbuffers/plugins/ruby:v21.2.0-1
     out: gen/proto/ruby
    # Building the Python build and building the mypy interfaces.
-  - remote: buf.build/protocolbuffers/plugins/python:v3.20.0-1
+  - remote: buf.build/protocolbuffers/plugins/python:v3.19.3-1
     out: gen/proto/python
   - remote: buf.build/grpc/plugins/python:v1.47.0-1
     out: gen/proto/python

dev/requirements.txt

@@ -50,7 +50,7 @@ black==22.6.0
 
 # Spark Connect (required)
 grpcio==1.48.1
-protobuf==4.21.6
+protobuf==3.19.4
 
 # Spark Connect python proto generation plugin (optional)
 mypy-protobuf==3.3.0

python/pyspark/sql/connect/proto/base_pb2.py

@@ -18,9 +18,10 @@
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
 # source: spark/connect/base.proto
 """Generated protocol buffer code."""
-from google.protobuf.internal import builder as _builder
 from google.protobuf import descriptor as _descriptor
 from google.protobuf import descriptor_pool as _descriptor_pool
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
 from google.protobuf import symbol_database as _symbol_database
 
 # @@protoc_insertion_point(imports)
@@ -38,8 +39,135 @@
     b'\n\x18spark/connect/base.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1cspark/connect/commands.proto\x1a\x1dspark/connect/relations.proto\x1a\x19spark/connect/types.proto"t\n\x04Plan\x12-\n\x04root\x18\x01 \x01(\x0b\x32\x17.spark.connect.RelationH\x00R\x04root\x12\x32\n\x07\x63ommand\x18\x02 \x01(\x0b\x32\x16.spark.connect.CommandH\x00R\x07\x63ommandB\t\n\x07op_type"\xc8\x02\n\x07Request\x12\x1b\n\tclient_id\x18\x01 \x01(\tR\x08\x63lientId\x12\x45\n\x0cuser_context\x18\x02 \x01(\x0b\x32".spark.connect.Request.UserContextR\x0buserContext\x12\'\n\x04plan\x18\x03 \x01(\x0b\x32\x13.spark.connect.PlanR\x04plan\x12$\n\x0b\x63lient_type\x18\x04 \x01(\tH\x00R\nclientType\x88\x01\x01\x1az\n\x0bUserContext\x12\x17\n\x07user_id\x18\x01 \x01(\tR\x06userId\x12\x1b\n\tuser_name\x18\x02 \x01(\tR\x08userName\x12\x35\n\nextensions\x18\xe7\x07 \x03(\x0b\x32\x14.google.protobuf.AnyR\nextensionsB\x0e\n\x0c_client_type"\xe0\x06\n\x08Response\x12\x1b\n\tclient_id\x18\x01 \x01(\tR\x08\x63lientId\x12\x45\n\x0b\x61rrow_batch\x18\x02 \x01(\x0b\x32".spark.connect.Response.ArrowBatchH\x00R\narrowBatch\x12\x42\n\njson_batch\x18\x03 \x01(\x0b\x32!.spark.connect.Response.JSONBatchH\x00R\tjsonBatch\x12\x39\n\x07metrics\x18\x04 \x01(\x0b\x32\x1f.spark.connect.Response.MetricsR\x07metrics\x1a=\n\nArrowBatch\x12\x1b\n\trow_count\x18\x01 \x01(\x03R\x08rowCount\x12\x12\n\x04\x64\x61ta\x18\x02 \x01(\x0cR\x04\x64\x61ta\x1a<\n\tJSONBatch\x12\x1b\n\trow_count\x18\x01 \x01(\x03R\x08rowCount\x12\x12\n\x04\x64\x61ta\x18\x02 \x01(\x0cR\x04\x64\x61ta\x1a\xe4\x03\n\x07Metrics\x12\x46\n\x07metrics\x18\x01 \x03(\x0b\x32,.spark.connect.Response.Metrics.MetricObjectR\x07metrics\x1a\xb6\x02\n\x0cMetricObject\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\x12\x17\n\x07plan_id\x18\x02 \x01(\x03R\x06planId\x12\x16\n\x06parent\x18\x03 \x01(\x03R\x06parent\x12o\n\x11\x65xecution_metrics\x18\x04 \x03(\x0b\x32\x42.spark.connect.Response.Metrics.MetricObject.ExecutionMetricsEntryR\x10\x65xecutionMetrics\x1ap\n\x15\x45xecutionMetricsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x41\n\x05value\x18\x02 \x01(\x0b\x32+.spark.connect.Response.Metrics.MetricValueR\x05value:\x02\x38\x01\x1aX\n\x0bMetricValue\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n\x05value\x18\x02 \x01(\x03R\x05value\x12\x1f\n\x0bmetric_type\x18\x03 \x01(\tR\nmetricTypeB\r\n\x0bresult_type"\x86\x01\n\x0f\x41nalyzeResponse\x12\x1b\n\tclient_id\x18\x01 \x01(\tR\x08\x63lientId\x12/\n\x06schema\x18\x02 \x01(\x0b\x32\x17.spark.connect.DataTypeR\x06schema\x12%\n\x0e\x65xplain_string\x18\x03 \x01(\tR\rexplainString2\xa2\x01\n\x13SparkConnectService\x12\x42\n\x0b\x45xecutePlan\x12\x16.spark.connect.Request\x1a\x17.spark.connect.Response"\x00\x30\x01\x12G\n\x0b\x41nalyzePlan\x12\x16.spark.connect.Request\x1a\x1e.spark.connect.AnalyzeResponse"\x00\x42"\n\x1eorg.apache.spark.connect.protoP\x01\x62\x06proto3'
 )
 
-_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
-_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "spark.connect.base_pb2", globals())
+
+_PLAN = DESCRIPTOR.message_types_by_name["Plan"]
+_REQUEST = DESCRIPTOR.message_types_by_name["Request"]
+_REQUEST_USERCONTEXT = _REQUEST.nested_types_by_name["UserContext"]
+_RESPONSE = DESCRIPTOR.message_types_by_name["Response"]
+_RESPONSE_ARROWBATCH = _RESPONSE.nested_types_by_name["ArrowBatch"]
+_RESPONSE_JSONBATCH = _RESPONSE.nested_types_by_name["JSONBatch"]
+_RESPONSE_METRICS = _RESPONSE.nested_types_by_name["Metrics"]
+_RESPONSE_METRICS_METRICOBJECT = _RESPONSE_METRICS.nested_types_by_name["MetricObject"]
+_RESPONSE_METRICS_METRICOBJECT_EXECUTIONMETRICSENTRY = (
+    _RESPONSE_METRICS_METRICOBJECT.nested_types_by_name["ExecutionMetricsEntry"]
+)
+_RESPONSE_METRICS_METRICVALUE = _RESPONSE_METRICS.nested_types_by_name["MetricValue"]
+_ANALYZERESPONSE = DESCRIPTOR.message_types_by_name["AnalyzeResponse"]
+Plan = _reflection.GeneratedProtocolMessageType(
+    "Plan",
+    (_message.Message,),
+    {
+        "DESCRIPTOR": _PLAN,
+        "__module__": "spark.connect.base_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.Plan)
+    },
+)
+_sym_db.RegisterMessage(Plan)
+
+Request = _reflection.GeneratedProtocolMessageType(
+    "Request",
+    (_message.Message,),
+    {
+        "UserContext": _reflection.GeneratedProtocolMessageType(
+            "UserContext",
+            (_message.Message,),
+            {
+                "DESCRIPTOR": _REQUEST_USERCONTEXT,
+                "__module__": "spark.connect.base_pb2"
+                # @@protoc_insertion_point(class_scope:spark.connect.Request.UserContext)
+            },
+        ),
+        "DESCRIPTOR": _REQUEST,
+        "__module__": "spark.connect.base_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.Request)
+    },
+)
+_sym_db.RegisterMessage(Request)
+_sym_db.RegisterMessage(Request.UserContext)
+
+Response = _reflection.GeneratedProtocolMessageType(
+    "Response",
+    (_message.Message,),
+    {
+        "ArrowBatch": _reflection.GeneratedProtocolMessageType(
+            "ArrowBatch",
+            (_message.Message,),
+            {
+                "DESCRIPTOR": _RESPONSE_ARROWBATCH,
+                "__module__": "spark.connect.base_pb2"
+                # @@protoc_insertion_point(class_scope:spark.connect.Response.ArrowBatch)
+            },
+        ),
+        "JSONBatch": _reflection.GeneratedProtocolMessageType(
+            "JSONBatch",
+            (_message.Message,),
+            {
+                "DESCRIPTOR": _RESPONSE_JSONBATCH,
+                "__module__": "spark.connect.base_pb2"
+                # @@protoc_insertion_point(class_scope:spark.connect.Response.JSONBatch)
+            },
+        ),
+        "Metrics": _reflection.GeneratedProtocolMessageType(
+            "Metrics",
+            (_message.Message,),
+            {
+                "MetricObject": _reflection.GeneratedProtocolMessageType(
+                    "MetricObject",
+                    (_message.Message,),
+                    {
+                        "ExecutionMetricsEntry": _reflection.GeneratedProtocolMessageType(
+                            "ExecutionMetricsEntry",
+                            (_message.Message,),
+                            {
+                                "DESCRIPTOR": _RESPONSE_METRICS_METRICOBJECT_EXECUTIONMETRICSENTRY,
+                                "__module__": "spark.connect.base_pb2"
+                                # @@protoc_insertion_point(class_scope:spark.connect.Response.Metrics.MetricObject.ExecutionMetricsEntry)
+                            },
+                        ),
+                        "DESCRIPTOR": _RESPONSE_METRICS_METRICOBJECT,
+                        "__module__": "spark.connect.base_pb2"
+                        # @@protoc_insertion_point(class_scope:spark.connect.Response.Metrics.MetricObject)
+                    },
+                ),
+                "MetricValue": _reflection.GeneratedProtocolMessageType(
+                    "MetricValue",
+                    (_message.Message,),
+                    {
+                        "DESCRIPTOR": _RESPONSE_METRICS_METRICVALUE,
+                        "__module__": "spark.connect.base_pb2"
+                        # @@protoc_insertion_point(class_scope:spark.connect.Response.Metrics.MetricValue)
+                    },
+                ),
+                "DESCRIPTOR": _RESPONSE_METRICS,
+                "__module__": "spark.connect.base_pb2"
+                # @@protoc_insertion_point(class_scope:spark.connect.Response.Metrics)
+            },
+        ),
+        "DESCRIPTOR": _RESPONSE,
+        "__module__": "spark.connect.base_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.Response)
+    },
+)
+_sym_db.RegisterMessage(Response)
+_sym_db.RegisterMessage(Response.ArrowBatch)
+_sym_db.RegisterMessage(Response.JSONBatch)
+_sym_db.RegisterMessage(Response.Metrics)
+_sym_db.RegisterMessage(Response.Metrics.MetricObject)
+_sym_db.RegisterMessage(Response.Metrics.MetricObject.ExecutionMetricsEntry)
+_sym_db.RegisterMessage(Response.Metrics.MetricValue)
+
+AnalyzeResponse = _reflection.GeneratedProtocolMessageType(
+    "AnalyzeResponse",
+    (_message.Message,),
+    {
+        "DESCRIPTOR": _ANALYZERESPONSE,
+        "__module__": "spark.connect.base_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.AnalyzeResponse)
+    },
+)
+_sym_db.RegisterMessage(AnalyzeResponse)
+
+_SPARKCONNECTSERVICE = DESCRIPTOR.services_by_name["SparkConnectService"]
 if _descriptor._USE_C_DESCRIPTORS == False:
 
     DESCRIPTOR._options = None

python/pyspark/sql/connect/proto/commands_pb2.py

@@ -18,9 +18,10 @@
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
 # source: spark/connect/commands.proto
 """Generated protocol buffer code."""
-from google.protobuf.internal import builder as _builder
 from google.protobuf import descriptor as _descriptor
 from google.protobuf import descriptor_pool as _descriptor_pool
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
 from google.protobuf import symbol_database as _symbol_database
 
 # @@protoc_insertion_point(imports)
@@ -36,8 +37,81 @@
     b'\n\x1cspark/connect/commands.proto\x12\rspark.connect\x1a\x1dspark/connect/relations.proto\x1a\x19spark/connect/types.proto"\x94\x02\n\x07\x43ommand\x12N\n\x0f\x63reate_function\x18\x01 \x01(\x0b\x32#.spark.connect.CreateScalarFunctionH\x00R\x0e\x63reateFunction\x12H\n\x0fwrite_operation\x18\x02 \x01(\x0b\x32\x1d.spark.connect.WriteOperationH\x00R\x0ewriteOperation\x12_\n\x15\x63reate_dataframe_view\x18\x03 \x01(\x0b\x32).spark.connect.CreateDataFrameViewCommandH\x00R\x13\x63reateDataframeViewB\x0e\n\x0c\x63ommand_type"\x97\x04\n\x14\x43reateScalarFunction\x12\x14\n\x05parts\x18\x01 \x03(\tR\x05parts\x12P\n\x08language\x18\x02 \x01(\x0e\x32\x34.spark.connect.CreateScalarFunction.FunctionLanguageR\x08language\x12\x1c\n\ttemporary\x18\x03 \x01(\x08R\ttemporary\x12>\n\x0e\x61rgument_types\x18\x04 \x03(\x0b\x32\x17.spark.connect.DataTypeR\rargumentTypes\x12\x38\n\x0breturn_type\x18\x05 \x01(\x0b\x32\x17.spark.connect.DataTypeR\nreturnType\x12\x31\n\x13serialized_function\x18\x06 \x01(\x0cH\x00R\x12serializedFunction\x12\'\n\x0eliteral_string\x18\x07 \x01(\tH\x00R\rliteralString"\x8b\x01\n\x10\x46unctionLanguage\x12!\n\x1d\x46UNCTION_LANGUAGE_UNSPECIFIED\x10\x00\x12\x19\n\x15\x46UNCTION_LANGUAGE_SQL\x10\x01\x12\x1c\n\x18\x46UNCTION_LANGUAGE_PYTHON\x10\x02\x12\x1b\n\x17\x46UNCTION_LANGUAGE_SCALA\x10\x03\x42\x15\n\x13\x66unction_definition"\x96\x01\n\x1a\x43reateDataFrameViewCommand\x12-\n\x05input\x18\x01 \x01(\x0b\x32\x17.spark.connect.RelationR\x05input\x12\x12\n\x04name\x18\x02 \x01(\tR\x04name\x12\x1b\n\tis_global\x18\x03 \x01(\x08R\x08isGlobal\x12\x18\n\x07replace\x18\x04 \x01(\x08R\x07replace"\xe6\x05\n\x0eWriteOperation\x12-\n\x05input\x18\x01 \x01(\x0b\x32\x17.spark.connect.RelationR\x05input\x12\x16\n\x06source\x18\x02 \x01(\tR\x06source\x12\x14\n\x04path\x18\x03 \x01(\tH\x00R\x04path\x12\x1f\n\ntable_name\x18\x04 \x01(\tH\x00R\ttableName\x12:\n\x04mode\x18\x05 \x01(\x0e\x32&.spark.connect.WriteOperation.SaveModeR\x04mode\x12*\n\x11sort_column_names\x18\x06 \x03(\tR\x0fsortColumnNames\x12\x31\n\x14partitioning_columns\x18\x07 \x03(\tR\x13partitioningColumns\x12\x43\n\tbucket_by\x18\x08 \x01(\x0b\x32&.spark.connect.WriteOperation.BucketByR\x08\x62ucketBy\x12\x44\n\x07options\x18\t \x03(\x0b\x32*.spark.connect.WriteOperation.OptionsEntryR\x07options\x1a:\n\x0cOptionsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1a[\n\x08\x42ucketBy\x12.\n\x13\x62ucket_column_names\x18\x01 \x03(\tR\x11\x62ucketColumnNames\x12\x1f\n\x0bnum_buckets\x18\x02 \x01(\x05R\nnumBuckets"\x89\x01\n\x08SaveMode\x12\x19\n\x15SAVE_MODE_UNSPECIFIED\x10\x00\x12\x14\n\x10SAVE_MODE_APPEND\x10\x01\x12\x17\n\x13SAVE_MODE_OVERWRITE\x10\x02\x12\x1d\n\x19SAVE_MODE_ERROR_IF_EXISTS\x10\x03\x12\x14\n\x10SAVE_MODE_IGNORE\x10\x04\x42\x0b\n\tsave_typeB"\n\x1eorg.apache.spark.connect.protoP\x01\x62\x06proto3'
 )
 
-_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
-_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "spark.connect.commands_pb2", globals())
+
+_COMMAND = DESCRIPTOR.message_types_by_name["Command"]
+_CREATESCALARFUNCTION = DESCRIPTOR.message_types_by_name["CreateScalarFunction"]
+_CREATEDATAFRAMEVIEWCOMMAND = DESCRIPTOR.message_types_by_name["CreateDataFrameViewCommand"]
+_WRITEOPERATION = DESCRIPTOR.message_types_by_name["WriteOperation"]
+_WRITEOPERATION_OPTIONSENTRY = _WRITEOPERATION.nested_types_by_name["OptionsEntry"]
+_WRITEOPERATION_BUCKETBY = _WRITEOPERATION.nested_types_by_name["BucketBy"]
+_CREATESCALARFUNCTION_FUNCTIONLANGUAGE = _CREATESCALARFUNCTION.enum_types_by_name[
+    "FunctionLanguage"
+]
+_WRITEOPERATION_SAVEMODE = _WRITEOPERATION.enum_types_by_name["SaveMode"]
+Command = _reflection.GeneratedProtocolMessageType(
+    "Command",
+    (_message.Message,),
+    {
+        "DESCRIPTOR": _COMMAND,
+        "__module__": "spark.connect.commands_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.Command)
+    },
+)
+_sym_db.RegisterMessage(Command)
+
+CreateScalarFunction = _reflection.GeneratedProtocolMessageType(
+    "CreateScalarFunction",
+    (_message.Message,),
+    {
+        "DESCRIPTOR": _CREATESCALARFUNCTION,
+        "__module__": "spark.connect.commands_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.CreateScalarFunction)
+    },
+)
+_sym_db.RegisterMessage(CreateScalarFunction)
+
+CreateDataFrameViewCommand = _reflection.GeneratedProtocolMessageType(
+    "CreateDataFrameViewCommand",
+    (_message.Message,),
+    {
+        "DESCRIPTOR": _CREATEDATAFRAMEVIEWCOMMAND,
+        "__module__": "spark.connect.commands_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.CreateDataFrameViewCommand)
+    },
+)
+_sym_db.RegisterMessage(CreateDataFrameViewCommand)
+
+WriteOperation = _reflection.GeneratedProtocolMessageType(
+    "WriteOperation",
+    (_message.Message,),
+    {
+        "OptionsEntry": _reflection.GeneratedProtocolMessageType(
+            "OptionsEntry",
+            (_message.Message,),
+            {
+                "DESCRIPTOR": _WRITEOPERATION_OPTIONSENTRY,
+                "__module__": "spark.connect.commands_pb2"
+                # @@protoc_insertion_point(class_scope:spark.connect.WriteOperation.OptionsEntry)
+            },
+        ),
+        "BucketBy": _reflection.GeneratedProtocolMessageType(
+            "BucketBy",
+            (_message.Message,),
+            {
+                "DESCRIPTOR": _WRITEOPERATION_BUCKETBY,
+                "__module__": "spark.connect.commands_pb2"
+                # @@protoc_insertion_point(class_scope:spark.connect.WriteOperation.BucketBy)
+            },
+        ),
+        "DESCRIPTOR": _WRITEOPERATION,
+        "__module__": "spark.connect.commands_pb2"
+        # @@protoc_insertion_point(class_scope:spark.connect.WriteOperation)
+    },
+)
+_sym_db.RegisterMessage(WriteOperation)
+_sym_db.RegisterMessage(WriteOperation.OptionsEntry)
+_sym_db.RegisterMessage(WriteOperation.BucketBy)
+
 if _descriptor._USE_C_DESCRIPTORS == False:
 
     DESCRIPTOR._options = None