fix

4paradigm · Jan 17, 2024 · 19e9199 · 19e9199
1 parent c788708
commit 19e9199
Show file tree

Hide file tree

Showing 5 changed files with 14 additions and 17 deletions.
diff --git a/java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/OpenmldbTable.java b/java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/OpenmldbTable.java
@@ -22,10 +22,8 @@
 import com._4paradigm.openmldb.sdk.SqlException;
 import com._4paradigm.openmldb.sdk.SqlExecutor;
 import com._4paradigm.openmldb.sdk.impl.SqlClusterExecutor;
-import com._4paradigm.openmldb.spark.read.OpenmldbReadConfig;
 import com._4paradigm.openmldb.spark.read.OpenmldbScanBuilder;
 import com._4paradigm.openmldb.spark.write.OpenmldbWriteBuilder;
-import com._4paradigm.openmldb.spark.write.OpenmldbWriteConfig;
 import org.apache.spark.sql.connector.catalog.SupportsRead;
 import org.apache.spark.sql.connector.catalog.SupportsWrite;
 import org.apache.spark.sql.connector.catalog.TableCapability;

diff --git a/...ctor/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbPartitionReaderFactory.java b/...ctor/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbPartitionReaderFactory.java
@@ -17,15 +17,16 @@
 
 package com._4paradigm.openmldb.spark.read;
 
+import com._4paradigm.openmldb.spark.OpenmldbConfig;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.connector.read.InputPartition;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.apache.spark.sql.connector.read.PartitionReaderFactory;
 
 public class OpenmldbPartitionReaderFactory implements PartitionReaderFactory {
-    private final OpenmldbReadConfig config;
+    private final OpenmldbConfig config;
 
-    public OpenmldbPartitionReaderFactory(OpenmldbReadConfig config) {
+    public OpenmldbPartitionReaderFactory(OpenmldbConfig config) {
         this.config = config;
     }
 

diff --git a/...enmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbScan.java b/...enmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbScan.java
@@ -17,16 +17,17 @@
 
 package com._4paradigm.openmldb.spark.read;
 
+import com._4paradigm.openmldb.spark.OpenmldbConfig;
 import org.apache.spark.sql.connector.read.Batch;
 import org.apache.spark.sql.connector.read.InputPartition;
 import org.apache.spark.sql.connector.read.PartitionReaderFactory;
 import org.apache.spark.sql.connector.read.Scan;
 import org.apache.spark.sql.types.StructType;
 
 public class OpenmldbScan implements Scan, Batch {
-    private final OpenmldbReadConfig config;
+    private final OpenmldbConfig config;
 
-    public OpenmldbScan(OpenmldbReadConfig config) {
+    public OpenmldbScan(OpenmldbConfig config) {
         this.config = config;
     }
 

diff --git a/...spark-connector/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbScanBuilder.java b/...spark-connector/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbScanBuilder.java
@@ -17,13 +17,14 @@
 
 package com._4paradigm.openmldb.spark.read;
 
+import com._4paradigm.openmldb.spark.OpenmldbConfig;
 import org.apache.spark.sql.connector.read.Scan;
 import org.apache.spark.sql.connector.read.ScanBuilder;
 
 public class OpenmldbScanBuilder implements ScanBuilder {
-    private final OpenmldbReadConfig config;
+    private final OpenmldbConfig config;
 
-    public OpenmldbScanBuilder(OpenmldbReadConfig config) {
+    public OpenmldbScanBuilder(OpenmldbConfig config) {
         this.config = config;
     }
 

diff --git a/...connector/src/main/scala/com/_4paradigm/openmldb/spark/read/OpenmldbPartitionReader.scala b/...connector/src/main/scala/com/_4paradigm/openmldb/spark/read/OpenmldbPartitionReader.scala
@@ -1,5 +1,6 @@
 package com._4paradigm.openmldb.spark.read
 
+import com._4paradigm.openmldb.spark.OpenmldbConfig
 import com._4paradigm.openmldb.sdk.{Schema, SdkOption}
 import com._4paradigm.openmldb.sdk.impl.SqlClusterExecutor
 import org.apache.spark.sql.catalyst.InternalRow
@@ -8,15 +9,10 @@ import org.apache.spark.unsafe.types.UTF8String
 
 import java.sql.Types
 
-class OpenmldbPartitionReader(config: OpenmldbReadConfig) extends PartitionReader[InternalRow] {
-
-  val option = new SdkOption
-  option.setZkCluster(config.zkCluster)
-  option.setZkPath(config.zkPath)
-  option.setLight(true)
-  val executor = new SqlClusterExecutor(option)
-  val dbName: String = config.dbName
-  val tableName: String = config.tableName
+class OpenmldbPartitionReader(config: OpenmldbConfig) extends PartitionReader[InternalRow] {
+  val executor = new SqlClusterExecutor(config.getSdkOption)
+  val dbName: String = config.getDB
+  val tableName: String = config.getTable
 
   val schema: Schema = executor.getTableSchema(dbName, tableName)
   executor.executeSQL(dbName, "SET @@execute_mode='online'")