[SPARK-52812][SQL] Make Spark Connect Catalog.createTable eager

python/pyspark/core/rdd.py

@@ -4843,7 +4843,8 @@ def sumApprox(
         jrdd = self.mapPartitions(lambda it: [float(sum(it))])._to_java_object_rdd()
         assert self.ctx._jvm is not None
         jdrdd = self.ctx._jvm.JavaDoubleRDD.fromRDD(jrdd.rdd())
-        r = jdrdd.sumApprox(timeout, confidence).getFinalValue()
+        partial = jdrdd.sumApprox(timeout, confidence)
+        r = partial.initialValue()
         return BoundedFloat(r.mean(), r.confidence(), r.low(), r.high())
 
     def meanApprox(

python/pyspark/tests/test_rdd.py

@@ -638,6 +638,18 @@ def test_distinct(self):
         self.assertEqual(result.getNumPartitions(), 5)
         self.assertEqual(result.count(), 3)
 
+    def test_count_approx_respects_timeout(self):
+        rdd = self.sc.range(1000000, numSlices=8)
+        start = time.time()
+        result = rdd.countApprox(timeout=100)
+        elapsed = time.time() - start
+        self.assertLess(elapsed, 10)
+        self.assertIsNotNone(result)
+
+    def test_count_approx_returns_exact_when_completed(self):
+        rdd = self.sc.parallelize(range(1000), 8)
+        self.assertEqual(rdd.countApprox(timeout=5000), 1000)
+
     def test_external_group_by_key(self):
         self.sc._conf.set("spark.python.worker.memory", "1m")
         N = 2000001

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/CatalogSuite.scala

@@ -151,6 +151,19 @@ class CatalogSuite extends ConnectFunSuite with RemoteSparkSession with SQLHelpe
     assert(spark.catalog.listTables().collect().isEmpty)
   }
 
+  test("createTable should be eager") {
+    val tableName = "eager_table"
+    withTable(tableName) {
+      withTempPath { dir =>
+        val session = spark
+        import session.implicits._
+        Seq((1, "a")).toDF("id", "value").write.parquet(dir.getPath)
+        spark.catalog.createTable(tableName, dir.getPath)
+        assert(spark.catalog.tableExists(tableName))
+      }
+    }
+  }
+
   test("Cache Table APIs") {
     val parquetTableName = "parquet_table"
     withTable(parquetTableName) {

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/Catalog.scala

@@ -484,7 +484,7 @@ class Catalog(sparkSession: SparkSession) extends catalog.Catalog {
       schema: StructType,
       description: String,
       options: Map[String, String]): DataFrame = {
-    sparkSession.newDataFrame { builder =>
+    val df = sparkSession.newDataFrame { builder =>
       val createTableBuilder = builder.getCatalogBuilder.getCreateTableBuilder
         .setTableName(tableName)
         .setSource(source)
@@ -494,6 +494,8 @@ class Catalog(sparkSession: SparkSession) extends catalog.Catalog {
         createTableBuilder.putOptions(k, v)
       }
     }
+    df.collect()
+    df
   }
 
   /**