-
Notifications
You must be signed in to change notification settings - Fork 393
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
68 changed files
with
1,098 additions
and
230 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
154 changes: 154 additions & 0 deletions
154
spark-connector/spark-connector-common/build.gradle.kts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,154 @@ | ||
| /* | ||
| * Copyright 2024 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
| plugins { | ||
| `maven-publish` | ||
| id("java") | ||
| id("idea") | ||
| alias(libs.plugins.shadow) | ||
| } | ||
|
|
||
| repositories { | ||
| mavenCentral() | ||
| } | ||
|
|
||
| val scalaVersion: String = project.properties["scalaVersion"] as? String ?: extra["defaultScalaVersion"].toString() | ||
| val sparkVersion: String = "3.3.4" | ||
| val sparkMajorVersion: String = sparkVersion.substringBeforeLast(".") | ||
| val icebergVersion: String = libs.versions.iceberg.get() | ||
| val kyuubiVersion: String = "1.7.4" | ||
| val scalaJava8CompatVersion: String = libs.versions.scala.java.compat.get() | ||
| val scalaCollectionCompatVersion: String = libs.versions.scala.collection.compat.get() | ||
|
|
||
| dependencies { | ||
| implementation(project(":catalogs:bundled-catalog", configuration = "shadow")) | ||
| implementation(project(":clients:client-java-runtime", configuration = "shadow")) | ||
| implementation(libs.guava) | ||
| compileOnly("org.apache.iceberg:iceberg-spark-runtime-${sparkMajorVersion}_$scalaVersion:$icebergVersion") | ||
| compileOnly("org.apache.kyuubi:kyuubi-spark-connector-hive_$scalaVersion:$kyuubiVersion") | ||
|
|
||
| compileOnly("org.apache.spark:spark-catalyst_$scalaVersion:$sparkVersion") | ||
| compileOnly("org.apache.spark:spark-core_$scalaVersion:$sparkVersion") | ||
| compileOnly("org.apache.spark:spark-sql_$scalaVersion:$sparkVersion") | ||
| compileOnly("org.scala-lang.modules:scala-java8-compat_$scalaVersion:$scalaJava8CompatVersion") | ||
|
|
||
| annotationProcessor(libs.lombok) | ||
| compileOnly(libs.lombok) | ||
|
|
||
| testAnnotationProcessor(libs.lombok) | ||
| testCompileOnly(libs.lombok) | ||
|
|
||
| testImplementation(project(":integration-test-common", "testArtifacts")) | ||
| testImplementation(project(":core")) { | ||
| // use log from spark, to avoid java.lang.NoSuchMethodError: org.apache.logging.slf4j.Log4jLoggerFactory: method <init>()V not found | ||
| exclude("org.slf4j") | ||
| exclude("org.apache.logging.log4j") | ||
| } | ||
| testImplementation(project(":server")) { | ||
| exclude("org.slf4j") | ||
| exclude("org.apache.logging.log4j") | ||
| } | ||
| testImplementation(project(":server-common")) { | ||
| exclude("org.slf4j") | ||
| exclude("org.apache.logging.log4j") | ||
| } | ||
|
|
||
| testImplementation(libs.hive2.common) { | ||
| exclude("org.apache.curator") | ||
| // use hadoop from Spark | ||
| exclude("org.apache.hadoop") | ||
| exclude("org.eclipse.jetty.aggregate", "jetty-all") | ||
| exclude("org.eclipse.jetty.orbit", "javax.servlet") | ||
| } | ||
| testImplementation(libs.hive2.metastore) { | ||
| exclude("co.cask.tephra") | ||
| exclude("com.github.joshelser") | ||
| exclude("com.google.code.findbugs", "jsr305") | ||
| exclude("com.google.code.findbugs", "sr305") | ||
| exclude("com.tdunning", "json") | ||
| exclude("com.zaxxer", "HikariCP") | ||
| exclude("io.dropwizard.metricss") | ||
| exclude("javax.transaction", "transaction-api") | ||
| exclude("org.apache.avro") | ||
| exclude("org.apache.curator") | ||
| exclude("org.apache.hbase") | ||
| exclude("org.apache.hadoop") | ||
| exclude("org.apache.parquet", "parquet-hadoop-bundle") | ||
| exclude("org.apache.zookeeper") | ||
| exclude("org.eclipse.jetty.aggregate", "jetty-all") | ||
| exclude("org.eclipse.jetty.orbit", "javax.servlet") | ||
| exclude("org.slf4j") | ||
| } | ||
| testImplementation(libs.junit.jupiter.api) | ||
| testImplementation(libs.junit.jupiter.params) | ||
| testImplementation(libs.mysql.driver) | ||
| testImplementation(libs.testcontainers) | ||
|
|
||
| testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion") | ||
| testImplementation("org.apache.iceberg:iceberg-hive-metastore:$icebergVersion") | ||
| testImplementation("org.apache.iceberg:iceberg-spark-runtime-${sparkMajorVersion}_$scalaVersion:$icebergVersion") | ||
| testImplementation("org.apache.kyuubi:kyuubi-spark-connector-hive_$scalaVersion:$kyuubiVersion") | ||
| // include spark-sql,spark-catalyst,hive-common,hdfs-client | ||
| testImplementation("org.apache.spark:spark-hive_$scalaVersion:$sparkVersion") { | ||
| // conflict with Gravitino server jersey | ||
| exclude("org.glassfish.jersey.core") | ||
| exclude("org.glassfish.jersey.containers") | ||
| exclude("org.glassfish.jersey.inject") | ||
| } | ||
| testImplementation("org.scala-lang.modules:scala-collection-compat_$scalaVersion:$scalaCollectionCompatVersion") | ||
|
|
||
| testRuntimeOnly(libs.junit.jupiter.engine) | ||
| } | ||
|
|
||
| tasks.test { | ||
| val skipUTs = project.hasProperty("skipTests") | ||
| if (skipUTs) { | ||
| // Only run integration tests | ||
| include("**/integration/**") | ||
| } | ||
|
|
||
| val testMode = project.properties["testMode"] as? String ?: "embedded" | ||
|
|
||
| val skipITs = project.hasProperty("skipITs") | ||
| if (skipITs || testMode == "embedded") { | ||
| // Exclude integration tests | ||
| exclude("**/integration/**") | ||
| } else { | ||
| dependsOn(tasks.jar) | ||
|
|
||
| doFirst { | ||
| environment("GRAVITINO_CI_HIVE_DOCKER_IMAGE", "datastrato/gravitino-ci-hive:0.1.10") | ||
| } | ||
|
|
||
| val init = project.extra.get("initIntegrationTest") as (Test) -> Unit | ||
| init(this) | ||
| } | ||
| } | ||
|
|
||
| tasks.clean { | ||
| delete("spark-warehouse") | ||
| } | ||
|
|
||
| val testJar by tasks.registering(Jar::class) { | ||
| archiveClassifier.set("tests") | ||
| from(sourceSets["test"].output) | ||
| } | ||
|
|
||
| configurations { | ||
| create("testArtifacts") | ||
| } | ||
|
|
||
| artifacts { | ||
| add("testArtifacts", testJar) | ||
| } | ||
|
|
||
| tasks.register<Copy>("copy") { | ||
| from(configurations.testRuntimeClasspath) | ||
| into("build/libs-runtime") | ||
| } | ||
|
|
||
| tasks.register<Copy>("copy2") { | ||
| from(configurations.testCompileClasspath) | ||
| into("build/libs-compile") | ||
| } |
File renamed without changes.
File renamed without changes.
File renamed without changes.
90 changes: 90 additions & 0 deletions
90
...mon/src/main/java/com/datastrato/gravitino/spark/connector/SparkTableChangeConverter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| /* | ||
| * Copyright 2024 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
| package com.datastrato.gravitino.spark.connector; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import org.apache.spark.sql.connector.catalog.TableChange; | ||
|
|
||
| public class SparkTableChangeConverter { | ||
| private SparkTypeConverter sparkTypeConverter; | ||
|
|
||
| public SparkTableChangeConverter(SparkTypeConverter sparkTypeConverter) { | ||
| this.sparkTypeConverter = sparkTypeConverter; | ||
| } | ||
|
|
||
| public com.datastrato.gravitino.rel.TableChange toGravitinoTableChange(TableChange change) { | ||
| if (change instanceof TableChange.SetProperty) { | ||
| TableChange.SetProperty setProperty = (TableChange.SetProperty) change; | ||
| return com.datastrato.gravitino.rel.TableChange.setProperty( | ||
| setProperty.property(), setProperty.value()); | ||
| } else if (change instanceof TableChange.RemoveProperty) { | ||
| TableChange.RemoveProperty removeProperty = (TableChange.RemoveProperty) change; | ||
| return com.datastrato.gravitino.rel.TableChange.removeProperty(removeProperty.property()); | ||
| } else if (change instanceof TableChange.AddColumn) { | ||
| TableChange.AddColumn addColumn = (TableChange.AddColumn) change; | ||
| return com.datastrato.gravitino.rel.TableChange.addColumn( | ||
| addColumn.fieldNames(), | ||
| sparkTypeConverter.toGravitinoType(addColumn.dataType()), | ||
| addColumn.comment(), | ||
| transformColumnPosition(addColumn.position()), | ||
| addColumn.isNullable()); | ||
| } else if (change instanceof TableChange.DeleteColumn) { | ||
| TableChange.DeleteColumn deleteColumn = (TableChange.DeleteColumn) change; | ||
| return com.datastrato.gravitino.rel.TableChange.deleteColumn( | ||
| deleteColumn.fieldNames(), deleteColumn.ifExists()); | ||
| } else if (change instanceof TableChange.UpdateColumnType) { | ||
| TableChange.UpdateColumnType updateColumnType = (TableChange.UpdateColumnType) change; | ||
| return com.datastrato.gravitino.rel.TableChange.updateColumnType( | ||
| updateColumnType.fieldNames(), | ||
| sparkTypeConverter.toGravitinoType(updateColumnType.newDataType())); | ||
| } else if (change instanceof TableChange.RenameColumn) { | ||
| TableChange.RenameColumn renameColumn = (TableChange.RenameColumn) change; | ||
| return com.datastrato.gravitino.rel.TableChange.renameColumn( | ||
| renameColumn.fieldNames(), renameColumn.newName()); | ||
| } else if (change instanceof TableChange.UpdateColumnPosition) { | ||
| TableChange.UpdateColumnPosition sparkUpdateColumnPosition = | ||
| (TableChange.UpdateColumnPosition) change; | ||
| com.datastrato.gravitino.rel.TableChange.UpdateColumnPosition gravitinoUpdateColumnPosition = | ||
| (com.datastrato.gravitino.rel.TableChange.UpdateColumnPosition) | ||
| com.datastrato.gravitino.rel.TableChange.updateColumnPosition( | ||
| sparkUpdateColumnPosition.fieldNames(), | ||
| transformColumnPosition(sparkUpdateColumnPosition.position())); | ||
| Preconditions.checkArgument( | ||
| !(gravitinoUpdateColumnPosition.getPosition() | ||
| instanceof com.datastrato.gravitino.rel.TableChange.Default), | ||
| "Doesn't support alter column position without specifying position"); | ||
| return gravitinoUpdateColumnPosition; | ||
| } else if (change instanceof TableChange.UpdateColumnComment) { | ||
| TableChange.UpdateColumnComment updateColumnComment = | ||
| (TableChange.UpdateColumnComment) change; | ||
| return com.datastrato.gravitino.rel.TableChange.updateColumnComment( | ||
| updateColumnComment.fieldNames(), updateColumnComment.newComment()); | ||
| } else if (change instanceof TableChange.UpdateColumnNullability) { | ||
| TableChange.UpdateColumnNullability updateColumnNullability = | ||
| (TableChange.UpdateColumnNullability) change; | ||
| return com.datastrato.gravitino.rel.TableChange.updateColumnNullability( | ||
| updateColumnNullability.fieldNames(), updateColumnNullability.nullable()); | ||
| } else { | ||
| throw new UnsupportedOperationException( | ||
| String.format("Unsupported table change %s", change.getClass().getName())); | ||
| } | ||
| } | ||
|
|
||
| private com.datastrato.gravitino.rel.TableChange.ColumnPosition transformColumnPosition( | ||
| TableChange.ColumnPosition columnPosition) { | ||
| if (null == columnPosition) { | ||
| return com.datastrato.gravitino.rel.TableChange.ColumnPosition.defaultPos(); | ||
| } else if (columnPosition instanceof TableChange.First) { | ||
| return com.datastrato.gravitino.rel.TableChange.ColumnPosition.first(); | ||
| } else if (columnPosition instanceof TableChange.After) { | ||
| TableChange.After after = (TableChange.After) columnPosition; | ||
| return com.datastrato.gravitino.rel.TableChange.ColumnPosition.after(after.column()); | ||
| } else { | ||
| throw new UnsupportedOperationException( | ||
| String.format( | ||
| "Unsupported table column position %s", columnPosition.getClass().getName())); | ||
| } | ||
| } | ||
| } |
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.