Refactor(spark-3.5): Make OpenHouse SQL extensions standalone

integrations/spark/spark-3.5/openhouse-spark-runtime/build.gradle

@@ -8,6 +8,11 @@ plugins {
 ext {
   icebergVersion = rootProject.ext.iceberg_1_5_version
   sparkVersion = '3.5.2'
+  // Antlr grammar sources are owned by this module (spark-3.5 keeps its own copy of the
+  // grammar so it can evolve independently of spark-3.1).
+  antlrPackageDirPrefix = "com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/"
+  antlrMainDir = "${projectDir}/src/main/antlr/${antlrPackageDirPrefix}"
+  antlrMainGeneratedSrcDir = "${project.buildDir}/generated-src/antlr/main/"
 }
 
 configurations {
@@ -16,18 +21,40 @@ configurations {
     exclude(group: 'org.mapstruct')
   }
   shadow.extendsFrom implementation
+
+  antlr
 }
 
 // Set source for antlr generated directory
 sourceSets {
   main {
     java {
-      srcDirs += "${project(':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12').buildDir}/generated-src/antlr/main"
+      srcDirs antlrMainGeneratedSrcDir
     }
   }
 }
 
+// Task to generate java sources using Antlr tool
+task runAntlr(type: JavaExec) {
+  inputs.dir antlrMainDir
+  outputs.dir antlrMainGeneratedSrcDir
+
+  mainClass = "org.antlr.v4.Tool"
+  args = ["${antlrMainDir}/OpenhouseSqlExtensions.g4",
+          "-visitor",
+          "-o", "${antlrMainGeneratedSrcDir}/${antlrPackageDirPrefix}",
+          "-package", "com.linkedin.openhouse.spark.sql.catalyst.parser.extensions"]
+  maxHeapSize = "64m"
+  classpath = configurations.antlr
+}
+
+compileJava.dependsOn runAntlr
+
 dependencies {
+  // Required because we remove antlr plugin dependencies from the compile configuration
+  runtimeOnly "org.antlr:antlr4-runtime:4.7.1"
+  antlr "org.antlr:antlr4:4.7.1"
+
   compileOnly(project(path: ':integrations:java:iceberg-1.5:openhouse-java-iceberg-1.5-runtime', configuration: 'shadow'))
   compileOnly("org.apache.spark:spark-hive_2.12:${sparkVersion}") {
     exclude group: 'org.apache.avro', module: 'avro'
@@ -52,12 +79,6 @@ dependencies {
   fatJarPackagedDependencies(project(path: ':integrations:java:iceberg-1.5:openhouse-java-iceberg-1.5-runtime', configuration: 'shadow')) {
     transitive = false
   }
-  fatJarPackagedDependencies(project(path: ':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12', configuration: 'shadow')) {
-    transitive = false
-  }
-  implementation(project(path: ':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12', configuration: 'shadow')) {
-    exclude group: "com.linkedin.iceberg", module: "iceberg-spark-runtime-3.1_2.12"
-  }
   implementation("com.linkedin.iceberg:iceberg-spark-runtime-3.5_2.12:" + icebergVersion)
 }
 

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4

@@ -0,0 +1,244 @@
+grammar OpenhouseSqlExtensions;
+
+@lexer::members {
+  /**
+   * This method will be called when we see '/*' and try to match it as a bracketed comment.
+   * If the next character is '+', it should be parsed as hint later, and we cannot match
+   * it as a bracketed comment.
+   *
+   * Returns true if the next character is '+'.
+   */
+  public boolean isHint() {
+    int nextChar = _input.LA(1);
+    if (nextChar == '+') {
+      return true;
+    } else {
+      return false;
+    }
+  }
+}
+
+singleStatement
+    : statement EOF
+    ;
+
+statement
+  : ALTER TABLE multipartIdentifier SET POLICY '(' retentionPolicy (columnRetentionPolicy)? ')'        #setRetentionPolicy
+  | ALTER TABLE multipartIdentifier SET POLICY '(' replicationPolicy ')'                               #setReplicationPolicy
+  | ALTER TABLE multipartIdentifier UNSET POLICY '(' replication ')'                                   #unSetReplicationPolicy
+  | ALTER TABLE multipartIdentifier SET POLICY '(' sharingPolicy ')'                                   #setSharingPolicy
+  | ALTER TABLE multipartIdentifier SET POLICY '(' historyPolicy ')'                                   #setHistoryPolicy
+  | ALTER TABLE multipartIdentifier MODIFY columnNameClause SET columnPolicy                           #setColumnPolicyTag
+  | GRANT privilege ON grantableResource TO principal                                                  #grantStatement
+  | REVOKE privilege ON grantableResource FROM principal                                               #revokeStatement
+  | SHOW GRANTS ON grantableResource                                                                   #showGrantsStatement
+  ;
+
+multipartIdentifier
+    : parts+=identifier ('.' parts+=identifier)*
+    ;
+
+privilege
+    : columnLevelPrivilege
+    | SELECT | DESCRIBE | ALTER | GRANT_REVOKE | CREATE_TABLE
+    ;
+
+columnLevelPrivilege
+    : SELECT policyTag
+    ;
+
+grantableResource
+    : TABLE multipartIdentifier
+    | DATABASE multipartIdentifier
+    ;
+
+principal
+    : identifier
+    ;
+
+identifier
+    : IDENTIFIER
+    | quotedIdentifier
+    | nonReserved
+    ;
+
+quotedIdentifier
+    : BACKQUOTED_IDENTIFIER
+    ;
+
+nonReserved
+    : ALTER | TABLE | SET | POLICY | RETENTION | SHARING | REPLICATION | HISTORY
+    | GRANT | REVOKE | ON | TO | SHOW | GRANTS | PATTERN | WHERE | COLUMN
+    ;
+
+sharingPolicy
+    : SHARING '=' BOOLEAN
+    ;
+
+BOOLEAN
+    : 'TRUE' | 'FALSE'
+    ;
+
+retentionPolicy
+    : RETENTION '=' duration
+    ;
+
+columnRetentionPolicy
+    : ON columnNameClause (columnRetentionPolicyPatternClause)?
+    ;
+
+replication
+    : REPLICATION
+    ;
+
+replicationPolicy
+    : replication '=' tableReplicationPolicy
+    ;
+
+tableReplicationPolicy
+    : '(' replicationPolicyClause (',' replicationPolicyClause)* ')'
+    ;
+
+replicationPolicyClause
+    : '{' replicationPolicyClusterClause (',' replicationPolicyIntervalClause)? '}'
+    ;
+
+replicationPolicyClusterClause
+    : DESTINATION ':' STRING
+    ;
+
+replicationPolicyIntervalClause
+    : INTERVAL ':' RETENTION_HOUR
+    | INTERVAL ':' RETENTION_DAY
+    ;
+
+columnRetentionPolicyPatternClause
+    : WHERE retentionColumnPatternClause
+    ;
+
+columnNameClause
+    : COLUMN identifier
+    ;
+
+retentionColumnPatternClause
+    : PATTERN '=' STRING
+    ;
+
+duration
+    : RETENTION_DAY
+    | RETENTION_YEAR
+    | RETENTION_MONTH
+    | RETENTION_HOUR
+    ;
+
+RETENTION_DAY
+    : POSITIVE_INTEGER 'D'
+    ;
+
+RETENTION_YEAR
+    : POSITIVE_INTEGER 'Y'
+    ;
+
+RETENTION_MONTH
+    : POSITIVE_INTEGER 'M'
+    ;
+
+RETENTION_HOUR
+    : POSITIVE_INTEGER 'H'
+    ;
+
+columnPolicy
+    : TAG '=' multiTagIdentifier
+    | TAG '=' '(' NONE ')'
+    ;
+
+multiTagIdentifier
+    : '(' policyTag (',' policyTag)* ')'
+    ;
+
+policyTag
+    : PII | HC
+    ;
+
+historyPolicy
+    : HISTORY maxAge? versions?
+    ;
+
+maxAge
+    : MAX_AGE'='duration
+    ;
+
+versions
+    : VERSIONS'='POSITIVE_INTEGER
+    ;
+
+ALTER: 'ALTER';
+TABLE: 'TABLE';
+SET: 'SET';
+UNSET: 'UNSET';
+POLICY: 'POLICY';
+RETENTION: 'RETENTION';
+REPLICATION: 'REPLICATION';
+HISTORY: 'HISTORY';
+SHARING: 'SHARING';
+GRANT: 'GRANT';
+REVOKE: 'REVOKE';
+ON: 'ON';
+TO: 'TO';
+FROM: 'FROM';
+SELECT: 'SELECT';
+DESCRIBE: 'DESCRIBE';
+GRANT_REVOKE: 'MANAGE GRANTS';
+CREATE_TABLE: 'CREATE TABLE';
+DATABASE: 'DATABASE';
+SHOW: 'SHOW';
+GRANTS: 'GRANTS';
+PATTERN: 'PATTERN';
+DESTINATION: 'DESTINATION';
+INTERVAL: 'INTERVAL';
+WHERE: 'WHERE';
+COLUMN: 'COLUMN';
+PII: 'PII';
+HC: 'HC';
+MODIFY: 'MODIFY';
+TAG: 'TAG';
+NONE: 'NONE';
+VERSIONS: 'VERSIONS';
+MAX_AGE: 'MAX_AGE';
+
+POSITIVE_INTEGER
+    : DIGIT+
+    ;
+
+STRING
+    : '\'' ( ~('\''|'\\') | ('\\' .) )* '\''
+    | '"' ( ~('"'|'\\') | ('\\' .) )* '"'
+    ;
+
+IDENTIFIER
+    : (LETTER | DIGIT | '_')+
+    ;
+
+BACKQUOTED_IDENTIFIER
+    : '`' ( ~'`' | '``' )* '`'
+    ;
+
+fragment DIGIT
+    : [0-9]
+    ;
+
+fragment LETTER
+    : [A-Z]
+    ;
+
+SIMPLE_COMMENT
+    : '--' ('\\\n' | ~[\r\n])* '\r'? '\n'? -> channel(HIDDEN)
+    ;
+
+BRACKETED_COMMENT
+    : '/*' {!isHint()}? (BRACKETED_COMMENT|.)*? '*/' -> channel(HIDDEN)
+    ;
+
+WS
+    : [ \r\n\t]+ -> channel(HIDDEN)
+    ;

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/java/com/linkedin/openhouse/spark/OpenHouseCatalog.java

@@ -0,0 +1,18 @@
+package com.linkedin.openhouse.spark;
+
+/**
+ * Catalog implementation to create, read, update and delete tables in OpenHouse. This class
+ * leverages Openhouse tableclient to perform CRUD operations on Tables resource in the Catalog
+ * service. This implementation provides client side catalog implementation for Iceberg tables in
+ * Spark.
+ *
+ * <p>Catalog can be instantiated as a Iceberg catalog, with following configurations:
+ * spark.sql.catalog.openhouse=org.apache.iceberg.spark.SparkCatalog
+ * spark.sql.catalog.openhouse.catalog-impl=com.linkedin.openhouse.spark.OpenHouseCatalog
+ * spark.sql.catalog.openhouse.metrics-reporter-impl=com.linkedin.openhouse.javaclient.OpenHouseMetricsReporter
+ * spark.sql.catalog.openhouse.uri=http://[openhouse service host]:[openhouse service port]
+ * spark.sql.catalog.openhouse.cluster=[openhouse cluster name]
+ *
+ * <p>It can be used in spark shell as follows: spark.sql("USE openhouse")
+ */
+public class OpenHouseCatalog extends com.linkedin.openhouse.javaclient.OpenHouseCatalog {}

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/extensions/OpenhouseSparkSessionExtensions.scala

@@ -0,0 +1,12 @@
+package com.linkedin.openhouse.spark.extensions
+
+import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseSparkSqlExtensionsParser
+import com.linkedin.openhouse.spark.sql.execution.datasources.v2.OpenhouseDataSourceV2Strategy
+import org.apache.spark.sql.SparkSessionExtensions
+
+class OpenhouseSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
+  override def apply(extensions: SparkSessionExtensions): Unit = {
+    extensions.injectParser { case (_, parser) => new OpenhouseSparkSqlExtensionsParser(parser) }
+    extensions.injectPlannerStrategy( spark => OpenhouseDataSourceV2Strategy(spark))
+  }
+}

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/constants/Principal.scala

@@ -0,0 +1,18 @@
+package com.linkedin.openhouse.spark.sql.catalyst.constants
+
+/**
+ * This object is used to represent keyword global user group "PUBLIC" which maps to the acl policy representation "*"
+ */
+object Principal {
+  private val GLOBAL_USER_GROUP = "PUBLIC"
+  private val GLOBAL_USER_GROUP_ACL = "*"
+  def apply(principal: String): String = principal toUpperCase() match {
+    case GLOBAL_USER_GROUP => GLOBAL_USER_GROUP_ACL
+    case _ => principal
+  }
+
+  def unapply(principalAcl: String): Option[String] = principalAcl match {
+    case GLOBAL_USER_GROUP_ACL => Some(GLOBAL_USER_GROUP)
+    case _ => Some(principalAcl)
+  }
+}

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/enums/GrantableResourceTypes.scala

@@ -0,0 +1,6 @@
+package com.linkedin.openhouse.spark.sql.catalyst.enums
+
+private[sql] object GrantableResourceTypes extends Enumeration {
+  type GrantableResourceType = Value
+  val TABLE, DATABASE = Value
+}