feat: Storage v2 binlog format for MilvusDataSource

build.sbt

@@ -1,8 +1,30 @@
 import scala.sys.process.Process
+import scala.io.Source
 
 import xerial.sbt.Sonatype._
 import Dependencies._
 
+// Load Sonatype Central credentials
+credentials += {
+  val credFile = Path.userHome / ".sbt" / "sonatype_central_credentials"
+  if (credFile.exists) {
+    val lines = Source.fromFile(credFile).getLines().toList
+    val props = lines.map { line =>
+      val parts = line.split("=", 2)
+      if (parts.length == 2) Some(parts(0).trim -> parts(1).trim) else None
+    }.flatten.toMap
+
+    Credentials(
+      "Sonatype Nexus Repository Manager",
+      props.getOrElse("host", "central.sonatype.com"),
+      props.getOrElse("user", ""),
+      props.getOrElse("password", "")
+    )
+  } else {
+    Credentials(Path.userHome / ".sbt" / "sonatype.credentials")
+  }
+}
+
 ThisBuild / organizationName := "zilliz"
 ThisBuild / organizationHomepage := Some(url("https://zilliz.com/"))
 // For cross-compiling (if applicable)
@@ -54,7 +76,7 @@ lazy val root = (project in file("."))
     assembly / parallelExecution := true,
     Test / parallelExecution := true,
     Compile / compile / parallelExecution := true,
-    version := "0.1.14-SNAPSHOT",
+    version := "0.2.1-SNAPSHOT",
     organization := "com.zilliz",
 
     // Fork JVM for run and tests to properly load native libraries
@@ -80,7 +102,11 @@ lazy val root = (project in file("."))
       "-Xss2m",
       "-Xmx4g",
       "-Djava.library.path=.",
-      "--add-opens=java.base/java.nio=ALL-UNNAMED"
+      "--add-opens=java.base/java.nio=ALL-UNNAMED",
+      "--add-opens=java.base/java.lang=ALL-UNNAMED",
+      "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED",
+      "--add-opens=java.base/java.util=ALL-UNNAMED",
+      "--add-opens=java.base/sun.security.action=ALL-UNNAMED"
     ),
 
     Test / envVars := Map(

scripts/spark_submit_demo.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+# Spark Submit Demo Script
+# This script demonstrates how to submit a Spark application using spark-submit
+
+set -e
+
+export LD_PRELOAD=$(pwd)/src/main/resources/native/libmilvus-storage.so && \
+    spark-submit \
+    --master "local[*]" \
+    --conf "spark.driver.extraJavaOptions=-Xss2m
+    -Djava.library.path=$(pwd)/src/main/resources/native
+    --add-opens=java.base/java.nio=ALL-UNNAMED" \
+    --conf "spark.driver.userClassPathFirst=true" \
+    --conf "spark.executor.userClassPathFirst=true" \
+    --class example.MilvusStorageMain \
+    target/scala-2.13/spark-connector-assembly-0.2.1-SNAPSHOT.jar

src/main/scala/MilvusClient.scala

@@ -34,6 +34,8 @@ import io.milvus.grpc.milvus.{
   DeleteRequest,
   DescribeCollectionRequest,
   DescribeCollectionResponse,
+  DropCollectionRequest,
+  FlushRequest,
   GetImportStateRequest,
   GetImportStateResponse,
   GetPersistentSegmentInfoRequest,
@@ -292,6 +294,51 @@ class MilvusClient(params: MilvusConnectionParams) {
     }
   }
 
+  def dropCollection(
+      dbName: String = "",
+      collectionName: String
+  ): Try[Status] = {
+    try {
+      val status = stub.dropCollection(
+        DropCollectionRequest(
+          dbName = dbName,
+          collectionName = collectionName
+        )
+      )
+      checkStatus("dropCollection", status)
+    } catch {
+      case e: Exception =>
+        Failure(
+          new Exception(s"Failed to drop collection: ${e.getMessage}")
+        )
+    }
+  }
+
+  def flush(
+      dbName: String = "",
+      collectionNames: Seq[String] = Seq.empty
+  ): Try[Status] = {
+    try {
+      val flushResponse = stub.flush(
+        FlushRequest(
+          dbName = dbName,
+          collectionNames = collectionNames
+        )
+      )
+      checkStatus("flush", flushResponse.status.getOrElse(
+        Status(
+          errorCode = ErrorCode.UnexpectedError,
+          reason = "Flush Status is empty"
+        )
+      ))
+    } catch {
+      case e: Exception =>
+        Failure(
+          new Exception(s"Failed to flush collection: ${e.getMessage}")
+        )
+    }
+  }
+
   def packFieldData(): FieldData = {
     FieldData(
       `type` = DataType.Int64,
@@ -560,7 +607,8 @@ class MilvusClient(params: MilvusConnectionParams) {
             partitionID = info.partitionID,
             numRows = info.numRows,
             state = info.state,
-            level = info.level
+            level = info.level,
+            storageVersion = info.storageVersion
           )
         )
       )
@@ -781,7 +829,8 @@ case class MilvusSegmentInfo(
     partitionID: Long,
     numRows: Long,
     state: SegmentState,
-    level: SegmentLevel
+    level: SegmentLevel,
+    storageVersion: Long = 1L
 )
 
 case class MilvusSegmentLogInfo(