wip

yliuuuu · Apr 22, 2024 · 11a94c0 · 11a94c0
1 parent 2bc246e
commit 11a94c0
Show file tree

Hide file tree

Showing 12 changed files with 276 additions and 8 deletions.
diff --git a/partiql-eval/src/main/kotlin/org/partiql/eval/PartiQLEngineDefault.kt b/partiql-eval/src/main/kotlin/org/partiql/eval/PartiQLEngineDefault.kt
@@ -3,6 +3,7 @@ package org.partiql.eval
 import org.partiql.eval.internal.Compiler
 import org.partiql.eval.internal.Environment
 import org.partiql.eval.internal.Symbols
+import org.partiql.eval.internal.operator.Operator
 import org.partiql.plan.PartiQLPlan
 import org.partiql.value.PartiQLValue
 import org.partiql.value.PartiQLValueExperimental
@@ -17,10 +18,19 @@ internal class PartiQLEngineDefault : PartiQLEngine {
             // 2. Compile with built symbols
             val compiler = Compiler(plan, session, symbols)
             val expression = compiler.compile()
-            return object : PartiQLStatement.Query {
-                override fun execute(): PartiQLValue {
-                    return expression.eval(Environment.empty)
+            when (expression) {
+                is Operator.Aggregation -> TODO("not possible")
+                is Operator.Ddl -> return object : PartiQLStatement.Ddl {
+                    override fun execute(): PartiQLValue {
+                        return expression.create()
+                    }
                 }
+                is Operator.Expr -> return object : PartiQLStatement.Query {
+                    override fun execute(): PartiQLValue {
+                        return expression.eval(Environment.empty)
+                    }
+                }
+                is Operator.Relation -> TODO("not possible")
             }
         } catch (ex: Exception) {
             // TODO wrap in some PartiQL Exception
@@ -37,6 +47,13 @@ internal class PartiQLEngineDefault : PartiQLEngine {
             } catch (ex: Exception) {
                 PartiQLResult.Error(ex)
             }
+
+            is PartiQLStatement.Ddl -> try {
+                val value = statement.execute()
+                PartiQLResult.Value(value)
+            } catch (ex: Exception) {
+                PartiQLResult.Error(ex)
+            }
         }
     }
 }
diff --git a/partiql-eval/src/main/kotlin/org/partiql/eval/PartiQLStatement.kt b/partiql-eval/src/main/kotlin/org/partiql/eval/PartiQLStatement.kt
@@ -12,4 +12,7 @@ sealed interface PartiQLStatement<T> {
 
     @OptIn(PartiQLValueExperimental::class)
     interface Query : PartiQLStatement<PartiQLValue>
+
+    @OptIn(PartiQLValueExperimental::class)
+    interface Ddl : PartiQLStatement<PartiQLValue>
 }
diff --git a/partiql-eval/src/main/kotlin/org/partiql/eval/internal/Compiler.kt b/partiql-eval/src/main/kotlin/org/partiql/eval/internal/Compiler.kt
@@ -2,6 +2,7 @@ package org.partiql.eval.internal
 
 import org.partiql.eval.PartiQLEngine
 import org.partiql.eval.internal.operator.Operator
+import org.partiql.eval.internal.operator.ddl.DdlCreate
 import org.partiql.eval.internal.operator.rel.RelAggregate
 import org.partiql.eval.internal.operator.rel.RelDistinct
 import org.partiql.eval.internal.operator.rel.RelExclude
@@ -40,6 +41,7 @@ import org.partiql.eval.internal.operator.rex.ExprTupleUnion
 import org.partiql.eval.internal.operator.rex.ExprVarLocal
 import org.partiql.eval.internal.operator.rex.ExprVarOuter
 import org.partiql.plan.Catalog
+import org.partiql.plan.DdlOp
 import org.partiql.plan.PartiQLPlan
 import org.partiql.plan.PlanNode
 import org.partiql.plan.Ref
@@ -61,7 +63,7 @@ internal class Compiler(
     private val symbols: Symbols
 ) : PlanBaseVisitor<Operator, StaticType?>() {
 
-    fun compile(): Operator.Expr {
+    fun compile(): Operator {
         return visitPartiQLPlan(plan, null)
     }
 
@@ -81,14 +83,27 @@ internal class Compiler(
         throw IllegalStateException(node.message)
     }
 
-    override fun visitPartiQLPlan(node: PartiQLPlan, ctx: StaticType?): Operator.Expr {
-        return visitStatement(node.statement, ctx) as Operator.Expr
+    override fun visitPartiQLPlan(node: PartiQLPlan, ctx: StaticType?): Operator {
+        return visitStatement(node.statement, ctx)
     }
 
     override fun visitStatementQuery(node: Statement.Query, ctx: StaticType?): Operator.Expr {
         return visitRex(node.root, ctx).modeHandled()
     }
 
+    override fun visitStatementDDL(node: Statement.DDL, ctx: StaticType?): Operator.Ddl {
+        return when (val op = node.op) {
+            is DdlOp.CreateTable -> DdlCreate(
+                op.name,
+                op.shape,
+                op.constraint,
+                op.partitionExpr,
+                op.tableProperties,
+                session.catalogs
+            )
+        }
+    }
+
     // REX
 
     override fun visitRex(node: Rex, ctx: StaticType?): Operator.Expr {

diff --git a/partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/Operator.kt b/partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/Operator.kt
@@ -42,4 +42,9 @@ internal sealed interface Operator {
             DISTINCT
         }
     }
+
+    interface Ddl : Operator {
+        @OptIn(PartiQLValueExperimental::class)
+        fun create(): PartiQLValue
+    }
 }
diff --git a/partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/ddl/DdlCreate.kt b/partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/ddl/DdlCreate.kt
@@ -0,0 +1,96 @@
+package org.partiql.eval.internal.operator.ddl
+
+import com.amazon.ionelement.api.field
+import com.amazon.ionelement.api.ionString
+import com.amazon.ionelement.api.ionStructOf
+import org.partiql.eval.internal.operator.Operator
+import org.partiql.plan.Constraint
+import org.partiql.plan.Identifier
+import org.partiql.plan.PartitionExpr
+import org.partiql.plan.TableProperty
+import org.partiql.spi.connector.Connector
+import org.partiql.spi.connector.ConnectorPath
+import org.partiql.spi.connector.ConnectorSession
+import org.partiql.types.BagType
+import org.partiql.types.StaticType
+import org.partiql.types.StructType
+import org.partiql.value.PartiQLValue
+import org.partiql.value.PartiQLValueExperimental
+import org.partiql.value.StringValue
+import org.partiql.value.int32Value
+
+internal class DdlCreate(
+    val name: Identifier,
+    val shape: StaticType,
+    val constraint: List<Constraint>,
+    val partitionExpr: PartitionExpr?,
+    val tableProperties: List<TableProperty>,
+    val catalogs: Map<String, Connector>
+) : Operator.Ddl {
+    @PartiQLValueExperimental
+    override fun create(): PartiQLValue {
+        val (prefix, tableName) = when(name) {
+            is Identifier.Qualified -> {
+                val list = mutableListOf<String>()
+                list.add(name.root.symbol)
+                while (list.size != name.steps.size) {
+                    list.add(name.steps[list.size - 1].symbol)
+                }
+                list.toList() to name.steps.last().symbol
+            }
+            is Identifier.Symbol -> emptyList<String>() to name.symbol
+        }
+
+        val session =  object : ConnectorSession {
+            override fun getQueryId(): String = "q"
+
+            override fun getUserId(): String = "u"
+        }
+
+        // if prefix is 0, how do we choose the default one ????
+        val catalog = when(prefix.size) {
+            0 -> catalogs.values.first()
+            1 -> catalogs.values.first()
+            else -> catalogs[prefix.first()] ?: error("no such connector")
+        }
+
+        val checkExpression = constraint
+            .filter { it.body is Constraint.Body.Check }
+            .map {
+                val body = it.body as Constraint.Body.Check
+                body.unlowered
+            }
+
+        val unique =  constraint
+            .filter { it.body is Constraint.Body.Unique && (it.body as Constraint.Body.Unique).isPrimaryKey != true }
+            .flatMap {
+                val body = it.body as Constraint.Body.Unique
+                body.columns.map { it.symbol }
+            }
+
+        val primaryKey = constraint
+            .filter { it.body is Constraint.Body.Unique && (it.body as Constraint.Body.Unique).isPrimaryKey != false }
+            .flatMap {
+                val body = it.body as Constraint.Body.Unique
+                body.columns.map { it.symbol }
+            }
+
+
+
+        return try {
+            catalog
+                .getMetadata(session)
+                .createTable(
+                ConnectorPath.of(*prefix.toTypedArray()),
+                tableName,
+                shape,
+                checkExpression,
+                unique,
+                primaryKey,
+            )
+            int32Value(1)
+        } catch (e: Exception) {
+            int32Value(-1)
+        }
+    }
+}
diff --git a/partiql-planner/src/main/kotlin/org/partiql/planner/internal/typer/PlanTyper.kt b/partiql-planner/src/main/kotlin/org/partiql/planner/internal/typer/PlanTyper.kt
@@ -1396,8 +1396,6 @@ internal class PlanTyper(private val env: Env) {
 
     private fun Statement.DDL.type() = DdlTyper().visitStatementDDL(this, ANY)
 
-    private fun rexErr(message: String) = rex(MISSING, rexOpErr(message))
-
     /**
      * I found decorating the tree with the binding names (for resolution) was easier than associating introduced
      * bindings with a node via an id->list<string> map. ONLY because right now I don't think we have a good way

diff --git a/partiql-spi/src/main/kotlin/org/partiql/spi/connector/ConnectorMetadata.kt b/partiql-spi/src/main/kotlin/org/partiql/spi/connector/ConnectorMetadata.kt
@@ -16,6 +16,7 @@ package org.partiql.spi.connector
 
 import org.partiql.spi.BindingPath
 import org.partiql.spi.fn.FnExperimental
+import org.partiql.types.StaticType
 
 /**
  * Aids in retrieving relevant Catalog metadata for the purpose of planning and execution.
@@ -55,4 +56,12 @@ public interface ConnectorMetadata {
      */
     @FnExperimental
     public fun getAggregation(path: BindingPath): ConnectorHandle.Agg?
+    public fun createTable(
+        path: ConnectorPath,
+        tableName: String,
+        shape: StaticType,
+        checkExpression: List<String>,
+        unique: List<String>,
+        primaryKey: List<String>
+    )
 }
diff --git a/partiql-spi/src/main/kotlin/org/partiql/spi/connector/sql/SqlMetadata.kt b/partiql-spi/src/main/kotlin/org/partiql/spi/connector/sql/SqlMetadata.kt
@@ -21,6 +21,7 @@ import org.partiql.spi.connector.ConnectorPath
 import org.partiql.spi.connector.ConnectorSession
 import org.partiql.spi.connector.sql.info.InfoSchema
 import org.partiql.spi.fn.FnExperimental
+import org.partiql.types.StaticType
 
 /**
  * An instance of [SqlMetadata]
@@ -62,4 +63,15 @@ public open class SqlMetadata(
         }
         return ConnectorHandle.Agg(ConnectorPath(cnf), SqlAgg(name, variants))
     }
+
+    override fun createTable(
+        path: ConnectorPath,
+        tableName: String,
+        shape: StaticType,
+        checkExpression: List<String>,
+        unique: List<String>,
+        primaryKey: List<String>
+    ) {
+        TODO("Not yet implemented")
+    }
 }
diff --git a/plugins/partiql-base-jdbc/src/main/kotlin/org/partiql/plugins/BaseJdbcClient.kt b/plugins/partiql-base-jdbc/src/main/kotlin/org/partiql/plugins/BaseJdbcClient.kt
@@ -105,6 +105,14 @@ public abstract class BaseJdbcClient(
         }
     }
 
+
+    override fun createTable(session: ConnectorSession, query: String) {
+        val connection = connectionFactory.openConnection(session) ?: error("Failed to establish connection")
+        val stmt = connection.createStatement();
+        stmt.executeUpdate(query);
+        stmt.close();
+    }
+
     private fun listSchema(connection: Connection): Set<String> {
         //  Single catalog; multiple schema; multiple tables under each schema
         val catalog = connection.catalog

diff --git a/plugins/partiql-base-jdbc/src/main/kotlin/org/partiql/plugins/BaseJdbcMetadata.kt b/plugins/partiql-base-jdbc/src/main/kotlin/org/partiql/plugins/BaseJdbcMetadata.kt
@@ -9,9 +9,27 @@ import org.partiql.spi.connector.ConnectorSession
 import org.partiql.spi.connector.sql.SqlMetadata
 import org.partiql.spi.connector.sql.info.InfoSchema
 import org.partiql.spi.fn.FnExperimental
+import org.partiql.types.AnyOfType
+import org.partiql.types.AnyType
 import org.partiql.types.BagType
+import org.partiql.types.BlobType
+import org.partiql.types.BoolType
+import org.partiql.types.ClobType
+import org.partiql.types.DateType
+import org.partiql.types.DecimalType
+import org.partiql.types.FloatType
+import org.partiql.types.GraphType
+import org.partiql.types.IntType
+import org.partiql.types.ListType
+import org.partiql.types.MissingType
+import org.partiql.types.NullType
+import org.partiql.types.SexpType
 import org.partiql.types.StaticType
+import org.partiql.types.StringType
 import org.partiql.types.StructType
+import org.partiql.types.SymbolType
+import org.partiql.types.TimeType
+import org.partiql.types.TimestampType
 import org.partiql.types.TupleConstraint
 import org.partiql.value.PartiQLValueExperimental
 import org.partiql.value.PartiQLValueType
@@ -83,6 +101,80 @@ public class BaseJdbcMetadata(
     override fun getAggregation(path: BindingPath): ConnectorHandle.Agg? =
         super.getAggregation(path)
 
+    override fun createTable(
+        path: ConnectorPath,
+        tableName: String,
+        shape: StaticType,
+        checkExpression: List<String>,
+        unique: List<String>,
+        primaryKey: List<String>
+    ) {
+        val query = buildString {
+            this.append("CREATE TABLE ")
+            path.steps.forEach {
+                this.append("$it.")
+            }
+            this.append(tableName)
+
+            this.appendLine("(")
+            val struct = (shape as BagType).elementType as StructType
+
+            struct.fields.forEach {
+                when(val v = it.value) {
+                    is AnyOfType -> TODO()
+                    is AnyType -> TODO()
+                    is BlobType -> TODO()
+                    is BoolType -> TODO()
+                    is ClobType -> TODO()
+                    is BagType -> TODO()
+                    is ListType -> TODO()
+                    is SexpType -> TODO()
+                    is DateType -> TODO()
+                    is DecimalType -> TODO()
+                    is FloatType -> TODO()
+                    is GraphType -> TODO()
+                    is IntType -> when(v.rangeConstraint) {
+                        IntType.IntRangeConstraint.SHORT -> this.appendLine("${it.key} INT2,")
+                        IntType.IntRangeConstraint.INT4 -> this.appendLine("${it.key} INT4,")
+                        IntType.IntRangeConstraint.LONG -> this.appendLine("${it.key} INT8,")
+                        IntType.IntRangeConstraint.UNCONSTRAINED -> TODO()
+                    }
+                    MissingType -> TODO()
+                    is NullType -> TODO()
+                    is StringType -> TODO()
+                    is StructType -> TODO()
+                    is SymbolType -> TODO()
+                    is TimeType -> TODO()
+                    is TimestampType -> TODO()
+                }
+            }
+
+            checkExpression.forEach {
+                this.appendLine("CHECK ($it),")
+            }
+            if (unique.size > 0) {
+                this.append("UNIQUE (")
+                unique.forEach {
+                    this.append("$it,")
+                }
+                this.setLength(this.length - 1)
+                this.appendLine(")")
+            }
+            if (primaryKey.size > 0) {
+                this.append("PRIMARY KEY (")
+                primaryKey.forEach {
+                    this.append("$it,")
+                }
+                this.setLength(this.length - 1)
+                this.appendLine(")")
+            }
+            this.setLength(this.length - 1)
+            this.appendLine(")")
+        }
+
+        jdbcClient.createTable(session, query)
+    }
+
     @OptIn(PartiQLValueExperimental::class)
     internal fun PartiQLValueType.toStaticType(): StaticType = when (this) {
         PartiQLValueType.NULL -> StaticType.NULL

diff --git a/plugins/partiql-base-jdbc/src/main/kotlin/org/partiql/plugins/jdbc/JdbcClient.kt b/plugins/partiql-base-jdbc/src/main/kotlin/org/partiql/plugins/jdbc/JdbcClient.kt
@@ -22,4 +22,6 @@ public interface JdbcClient {
     public fun toColumnMapping(session: ConnectorSession, connection: Connection, jdbcType: JdbcType): ColumnMapping
 
     public fun getBinding(session: ConnectorSession) : ConnectorBindings
+
+    public fun createTable(session: ConnectorSession, query: String)
 }