Feature/265 deterministic plan (#266)

* Use static uuid for new columns

Signed-off-by: Kevin Wallimann <[email protected]>

* Use static uuid for new columns

Signed-off-by: Kevin Wallimann <[email protected]>

* Use maven 3.8.4

Signed-off-by: Kevin Wallimann <[email protected]>

* Use maven 3.8.4

Signed-off-by: Kevin Wallimann <[email protected]>

Author: kevinwallimann

.github/workflows/build.yml

@@ -28,9 +28,11 @@ jobs:
           key: ${{ runner.os }}-${{ matrix.scala }}-${{ matrix.spark }}-${{ hashFiles('**/pom.xml') }}
           restore-keys: |
             ${{ runner.os }}-${{ matrix.scala }}-${{ matrix.spark }}-
+      - name: Install Maven 3.8.4
+        run: mvn -N wrapper:wrapper -Dmaven=3.8.4
       - name: License check
-        run: mvn clean validate --no-transfer-progress -Plicense-check,spark-2,spark-3,scala-2.12
+        run: ./mvnw clean validate --no-transfer-progress -Plicense-check,spark-2,spark-3,scala-2.12
       - name: Switch scala version
-        run: mvn scala-cross-build:change-version --no-transfer-progress -Pscala-${{ matrix.scala }},spark-${{ matrix.spark }}
+        run: ./mvnw scala-cross-build:change-version --no-transfer-progress -Pscala-${{ matrix.scala }},spark-${{ matrix.spark }}
       - name: Build and run tests
-        run: mvn clean verify --no-transfer-progress -Pscala-${{ matrix.scala }},spark-${{ matrix.spark }},all-tests
+        run: ./mvnw clean verify --no-transfer-progress -Pscala-${{ matrix.scala }},spark-${{ matrix.spark }},all-tests

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/ConfluentAvroDecodingTransformer.scala

@@ -15,7 +15,6 @@
 
 package za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent
 
-import java.util.UUID
 import org.apache.commons.configuration2.Configuration
 import org.apache.commons.lang3.RandomStringUtils
 import org.slf4j.LoggerFactory
@@ -27,10 +26,10 @@ import za.co.absa.abris.config.FromAvroConfig
 import za.co.absa.hyperdrive.ingestor.api.context.HyperdriveContext
 import za.co.absa.hyperdrive.ingestor.api.transformer.{StreamTransformer, StreamTransformerFactory}
 import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils
-import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils.getOrThrow
 import za.co.absa.hyperdrive.ingestor.implementation.HyperdriveContextKeys
 import za.co.absa.hyperdrive.ingestor.implementation.reader.kafka.KafkaStreamReader.KEY_TOPIC
-import za.co.absa.hyperdrive.ingestor.implementation.utils.{AbrisConfigKeys, AbrisConfigUtil, AbrisConsumerConfigKeys, SchemaRegistryConfigUtil}
+import za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent.ConfluentAvroDecodingTransformer.ColumnPrefix
+import za.co.absa.hyperdrive.ingestor.implementation.utils.{AbrisConfigUtil, AbrisConsumerConfigKeys, SchemaRegistryConfigUtil}
 
 private[transformer] class ConfluentAvroDecodingTransformer(
   val valueAvroConfig: FromAvroConfig,
@@ -48,8 +47,8 @@ private[transformer] class ConfluentAvroDecodingTransformer(
   }
 
   private def getKeyValueDataFrame(dataFrame: DataFrame, keyAvroConfig: FromAvroConfig) = {
-    val keyStructCol = UUID.randomUUID().toString
-    val valueStructCol = UUID.randomUUID().toString
+    val keyStructCol = ColumnPrefix + "-key"
+    val valueStructCol = ColumnPrefix + "-value"
     val columnsToSelect = Seq(
       from_avro(col("key"), keyAvroConfig) as Symbol(keyStructCol),
       from_avro(col("value"), valueAvroConfig) as Symbol(valueStructCol)
@@ -73,7 +72,7 @@ private[transformer] class ConfluentAvroDecodingTransformer(
   }
 
   private def getValueDataFrame(dataFrame: DataFrame) = {
-    val dataStructCol = UUID.randomUUID().toString
+    val dataStructCol = ColumnPrefix
     val columnsToSelect = Seq(
       from_avro(col("value"), valueAvroConfig) as Symbol(dataStructCol)
     ) ++ keepColumns.map(col)
@@ -85,7 +84,7 @@ private[transformer] class ConfluentAvroDecodingTransformer(
     nonNullableDf.select(dataColumns ++ keepColumns.map(col):_*)
   }
 
-  private def checkIfColumnNameConflictsExistWithKeepColumns(avroColumns: Seq[String]) = {
+  private def checkIfColumnNameConflictsExistWithKeepColumns(avroColumns: Seq[String]): Unit = {
     val nameCollisions = avroColumns.intersect(keepColumns)
     if (nameCollisions.nonEmpty) {
       throw new IllegalArgumentException(s"Names of columns to keep collided with key and value columns. Consider renaming them before. Conflicts: $nameCollisions")
@@ -107,6 +106,8 @@ private[transformer] class ConfluentAvroDecodingTransformer(
 object ConfluentAvroDecodingTransformer extends StreamTransformerFactory with ConfluentAvroDecodingTransformerAttributes {
   private val keyColumnPrefixLength = 4
 
+  val ColumnPrefix = "hyperdrive-22c9fda5-d56c-44e6-9c5f-13197e71f3fc"
+
   object AbrisConfigKeys extends AbrisConsumerConfigKeys {
     override val topic: String = KEY_TOPIC
     override val schemaId: String = KEY_SCHEMA_REGISTRY_VALUE_SCHEMA_ID

pom.xml

@@ -284,6 +284,9 @@
                                 <exclude>.editorconfig</exclude>
                                 <exclude>**/*.MockMaker</exclude>
                                 <exclude>**/*.json</exclude>
+                                <exclude>mvnw</exclude>
+                                <exclude>mvnw.cmd</exclude>
+                                <exclude>.mvn/**/*</exclude>
                             </excludes>
                         </configuration>
                     </plugin>