Merge pull request #49 from sauliusvl/uuid7

Use UUIDv7 based Iceberg file names

Author: shivam247

build.sbt

@@ -34,13 +34,13 @@ ThisBuild / git.remoteRepo := {
   }
 }
 
-val scalaTestVersion = "3.2.18"
+val scalaTestVersion = "3.2.19"
 val scalaCheckVersion = "1.18.0"
-val scalaCheckTestVersion = "3.2.18.0"
+val scalaCheckTestVersion = "3.2.19.0"
 
 val hadoopVersion = "3.4.0"
 val parquetVersion = "1.14.1"
-val icebergVersion = "1.5.2"
+val icebergVersion = "1.6.0"
 
 lazy val `stream-loader-core` = project
   .in(file("stream-loader-core"))
@@ -51,17 +51,17 @@ lazy val `stream-loader-core` = project
     buildInfoKeys := Seq[BuildInfoKey](name, version, scalaVersion, git.gitHeadCommit),
     libraryDependencies ++= Seq(
       "org.scala-lang"     % "scala-reflect"     % scalaVersion.value,
-      "org.apache.kafka"   % "kafka-clients"     % "3.7.0",
+      "org.apache.kafka"   % "kafka-clients"     % "3.8.0",
       "org.log4s"         %% "log4s"             % "1.10.0",
       "org.apache.commons" % "commons-compress"  % "1.26.2",
       "org.xerial.snappy"  % "snappy-java"       % "1.1.10.5",
       "org.lz4"            % "lz4-java"          % "1.8.0",
-      "com.github.luben"   % "zstd-jni"          % "1.5.6-3",
+      "com.github.luben"   % "zstd-jni"          % "1.5.6-4",
       "com.univocity"      % "univocity-parsers" % "2.9.1",
       "org.json4s"        %% "json4s-native"     % "4.0.7",
-      "io.micrometer"      % "micrometer-core"   % "1.13.1",
+      "io.micrometer"      % "micrometer-core"   % "1.13.2",
       "org.scalatest"     %% "scalatest"         % scalaTestVersion      % "test",
-      "org.scalatestplus" %% "scalacheck-1-17"   % scalaCheckTestVersion % "test",
+      "org.scalatestplus" %% "scalacheck-1-18"   % scalaCheckTestVersion % "test",
       "org.scalacheck"    %% "scalacheck"        % scalaCheckVersion     % "test",
       "ch.qos.logback"     % "logback-classic"   % "1.5.6"               % "test"
     )
@@ -75,9 +75,9 @@ lazy val `stream-loader-clickhouse` = project
     resolvers += "jitpack" at "https://jitpack.io",
     libraryDependencies ++= Seq(
       "org.apache.httpcomponents.client5" % "httpclient5"     % "5.3.1",
-      "com.clickhouse"                    % "clickhouse-jdbc" % "0.6.1",
+      "com.clickhouse"                    % "clickhouse-jdbc" % "0.6.3",
       "org.scalatest"                    %% "scalatest"       % scalaTestVersion      % "test",
-      "org.scalatestplus"                %% "scalacheck-1-17" % scalaCheckTestVersion % "test",
+      "org.scalatestplus"                %% "scalacheck-1-18" % scalaCheckTestVersion % "test",
       "org.scalacheck"                   %% "scalacheck"      % scalaCheckVersion     % "test"
     )
   )
@@ -116,14 +116,14 @@ lazy val `stream-loader-s3` = project
   .settings(commonSettings)
   .settings(
     libraryDependencies ++= Seq(
-      "software.amazon.awssdk" % "s3"              % "2.26.3",
+      "software.amazon.awssdk" % "s3"              % "2.26.25",
       "org.scalatest"         %% "scalatest"       % scalaTestVersion % "test",
-      "com.amazonaws"          % "aws-java-sdk-s3" % "1.12.744"       % "test",
+      "com.amazonaws"          % "aws-java-sdk-s3" % "1.12.765"       % "test",
       "org.gaul"               % "s3proxy"         % "2.2.0"          % "test"
     )
   )
 
-val verticaVersion = "24.2.0-0"
+val verticaVersion = "24.3.0-0"
 
 lazy val `stream-loader-vertica` = project
   .in(file("stream-loader-vertica"))
@@ -133,7 +133,7 @@ lazy val `stream-loader-vertica` = project
     libraryDependencies ++= Seq(
       "com.vertica.jdbc"   % "vertica-jdbc"    % verticaVersion        % "provided",
       "org.scalatest"     %% "scalatest"       % scalaTestVersion      % "test",
-      "org.scalatestplus" %% "scalacheck-1-17" % scalaCheckTestVersion % "test",
+      "org.scalatestplus" %% "scalacheck-1-18" % scalaCheckTestVersion % "test",
       "org.scalacheck"    %% "scalacheck"      % scalaCheckVersion     % "test"
     )
   )
@@ -167,11 +167,11 @@ lazy val `stream-loader-tests` = project
       "com.vertica.jdbc"                 % "vertica-jdbc"                     % verticaVersion,
       "org.scalacheck"                  %% "scalacheck"                       % scalaCheckVersion,
       "org.scalatest"                   %% "scalatest"                        % scalaTestVersion      % "test",
-      "org.scalatestplus"               %% "scalacheck-1-17"                  % scalaCheckTestVersion % "test",
+      "org.scalatestplus"               %% "scalacheck-1-18"                  % scalaCheckTestVersion % "test",
       "org.slf4j"                        % "log4j-over-slf4j"                 % "2.0.13"              % "test",
       "org.mandas"                       % "docker-client"                    % "7.0.8"               % "test",
       "org.jboss.resteasy"               % "resteasy-client"                  % "6.2.9.Final"         % "test",
-      "com.fasterxml.jackson.jakarta.rs" % "jackson-jakarta-rs-json-provider" % "2.17.1"              % "test",
+      "com.fasterxml.jackson.jakarta.rs" % "jackson-jakarta-rs-json-provider" % "2.17.2"              % "test",
       "org.duckdb"                       % "duckdb_jdbc"                      % duckdbVersion         % "test"
     ),
     inConfig(IntegrationTest)(Defaults.testTasks),

project/build.properties

@@ -1 +1 @@
-sbt.version=1.10.0
+sbt.version=1.10.1

project/plugins.sbt

@@ -16,10 +16,10 @@ addSbtPlugin("com.github.sbt" % "sbt-unidoc" % "0.5.0")
 
 addSbtPlugin("de.heikoseeberger" % "sbt-header" % "5.10.0")
 
-libraryDependencies += "net.sourceforge.plantuml" % "plantuml" % "1.2024.5"
+libraryDependencies += "net.sourceforge.plantuml" % "plantuml" % "1.2024.6"
 
 addSbtPlugin("com.github.sbt" % "sbt-ghpages" % "0.8.0")
 
 addSbtPlugin("com.github.sbt" % "sbt-pgp" % "2.2.1")
 
-addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "3.10.0")
+addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "3.11.1")

stream-loader-tests/src/main/scala/com/adform/streamloader/util/UuidExtensions.scala stream-loader-core/src/main/scala/com/adform/streamloader/util/UuidExtensions.scala

@@ -9,6 +9,7 @@
 package com.adform.streamloader.util
 
 import java.nio.ByteBuffer
+import java.security.SecureRandom
 import java.util.UUID
 
 object UuidExtensions {
@@ -20,4 +21,24 @@ object UuidExtensions {
       bb.array
     }
   }
+
+  private val random = new SecureRandom
+
+  def randomUUIDv7(): UUID = {
+    val value = new Array[Byte](16)
+    random.nextBytes(value)
+
+    val timestamp = ByteBuffer.allocate(8)
+    timestamp.putLong(System.currentTimeMillis)
+    System.arraycopy(timestamp.array, 2, value, 0, 6)
+
+    value(6) = ((value(6) & 0x0f) | 0x70).toByte
+    value(8) = ((value(8) & 0x3f) | 0x80).toByte
+
+    val buf = ByteBuffer.wrap(value)
+    val high = buf.getLong
+    val low = buf.getLong
+
+    new UUID(high, low)
+  }
 }

stream-loader-core/src/test/scala/com/adform/streamloader/util/UuidExtensionsTest.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2020 Adform
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+package com.adform.streamloader.util
+
+import com.adform.streamloader.util.UuidExtensions.randomUUIDv7
+import org.scalatest.funspec.AnyFunSpec
+import org.scalatest.matchers.should.Matchers
+
+class UuidExtensionsTest extends AnyFunSpec with Matchers {
+
+  it("should generate unique UUIDv7 values") {
+    val uuids = (0 until 100).map(_ => randomUUIDv7())
+    uuids should contain theSameElementsInOrderAs uuids.distinct
+  }
+
+  it("should generate alphabetically sorted UUIDv7 values") {
+    val uuids = (0 until 10)
+      .map(_ => {
+        randomUUIDv7()
+        Thread.sleep(5) // the v7 spec guarantees ordering in a 1ms scope
+      })
+      .map(_.toString)
+
+    uuids should contain theSameElementsInOrderAs uuids.sorted
+  }
+}

stream-loader-iceberg/src/main/scala/com/adform/streamloader/iceberg/IcebergRecordBatcher.scala

@@ -12,12 +12,12 @@ import com.adform.streamloader.model.{StreamRange, StreamRecord}
 import com.adform.streamloader.sink.batch.{RecordBatch, RecordBatchBuilder, RecordBatcher, RecordFormatter}
 import com.adform.streamloader.sink.file.{FileStats, MultiFileCommitStrategy}
 import com.adform.streamloader.util.TimeProvider
+import com.adform.streamloader.util.UuidExtensions.randomUUIDv7
 import org.apache.iceberg.data.{GenericAppenderFactory, Record => IcebergRecord}
 import org.apache.iceberg.io.DataWriteResult
 import org.apache.iceberg.{FileFormat, PartitionKey, Table}
 
 import java.time.Duration
-import java.util.UUID
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
 
@@ -41,7 +41,7 @@ class IcebergRecordBatchBuilder(
     private val startTimeMillis = timeProvider.currentMillis
     private var recordCount = 0L
     private val dataWriter = {
-      val filename = fileFormat.addExtension(UUID.randomUUID().toString)
+      val filename = fileFormat.addExtension(randomUUIDv7().toString)
       val path = table.locationProvider().newDataLocation(table.spec(), pk, filename)
       val output = table.io().newOutputFile(path)