Pushdown/join

README-template.md

@@ -179,6 +179,7 @@ table|String|The Table of the Cloud Spanner database that you are reading from
 enableDataboost|Boolean|Enable the [Data Boost](https://cloud.google.com/spanner/docs/databoost/databoost-overview), which provides independent compute resources to query Spanner with near-zero impact to existing workloads. Note1: the option may trigger [extra charge](https://cloud.google.com/spanner/pricing#spanner-data-boost-pricing). Note2: this feature is not supported in Spanner Omni.
 readTimestamp|String|An RFC 3339 timestamp identifying the database snapshot to read. Multiple table reads using the same timestamp observe a consistent snapshot. By default, each table read uses the time when its scan is created.
 emulatorHost|String|The host and port of the Spanner emulator (e.g. `localhost:9010`) or a Spanner Omni instance (e.g. `localhost:15000`). When set, the connector connects to the emulator or Spanner Omni instead of Cloud Spanner. Useful for local development and testing.
+enablePredicateSql|Boolean|Enable SQL generation for Spark Predicates
 
 ### Writing to Spanner Tables
 

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerInformationSchema.java

@@ -78,6 +78,16 @@ default String dropTableDdl(String tableName) {
     return "DROP TABLE " + quoteIdentifier(tableName);
   }
 
+  static Statement getInterleaving() {
+    return Statement.of(
+        "SELECT\n"
+            + "    table_name,\n"
+            + "    parent_table_name,\n"
+            + "    interleave_type\n"
+            + "FROM information_schema.tables\n"
+            + "WHERE table_schema = ''");
+  }
+
   static SpannerInformationSchema create(Dialect dialect) {
     switch (dialect) {
       case POSTGRESQL:

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SparkSpannerTableProviderBase.java

@@ -58,7 +58,7 @@ public Table getTable(
     boolean hasTable = options.containsKey("table");
     boolean hasGraph = options.containsKey("graph");
     if (hasTable && !hasGraph) {
-      return new SpannerTable(options, schema);
+      return createSpannerTable(options, schema);
     } else if (!hasTable && hasGraph) {
       return SpannerGraphBuilder.build(options);
     } else {
@@ -68,6 +68,10 @@ public Table getTable(
     }
   }
 
+  public SpannerTable createSpannerTable(CaseInsensitiveStringMap options, StructType schema) {
+    return new SpannerTable(options, schema);
+  }
+
   /*
    * Returns true if the source has the ability of
    * accepting external table metadata when getting tables.

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/binding/GoogleSqlParameterRegistry.java

@@ -0,0 +1,10 @@
+package com.google.cloud.spark.spanner.binding;
+
+public class GoogleSqlParameterRegistry extends ParameterRegistry {
+  @Override
+  public ParameterRef nextParameter() {
+    counter++;
+    final String name = "p" + String.valueOf(counter);
+    return new ParameterRef(name, name);
+  }
+}

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/binding/ParameterRef.java

@@ -0,0 +1,19 @@
+package com.google.cloud.spark.spanner.binding;
+
+public class ParameterRef {
+  private final String bindName;
+  private final String sqlName;
+
+  public ParameterRef(String bindName, String sqlName) {
+    this.bindName = bindName;
+    this.sqlName = sqlName;
+  }
+
+  public String getBindName() {
+    return bindName;
+  }
+
+  public String getSqlName() {
+    return sqlName;
+  }
+}

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/binding/ParameterRegistry.java

@@ -0,0 +1,19 @@
+package com.google.cloud.spark.spanner.binding;
+
+import com.google.cloud.spanner.Dialect;
+
+public abstract class ParameterRegistry {
+  protected int counter = 0;
+
+  public abstract ParameterRef nextParameter();
+
+  public static ParameterRegistry create(Dialect dialect) {
+    switch (dialect) {
+      case POSTGRESQL:
+        return new PostgresSqlParameterRegistry();
+      case GOOGLE_STANDARD_SQL:
+        return new GoogleSqlParameterRegistry();
+    }
+    throw new IllegalArgumentException("Unsupported dialect: " + dialect);
+  }
+}

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/binding/PostgresSqlParameterRegistry.java

@@ -0,0 +1,10 @@
+package com.google.cloud.spark.spanner.binding;
+
+public class PostgresSqlParameterRegistry extends ParameterRegistry {
+  @Override
+  public ParameterRef nextParameter() {
+    counter++;
+    final String name = String.valueOf(counter);
+    return new ParameterRef("p" + name, name);
+  }
+}

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/binding/SpannerTypeBinder.java

@@ -0,0 +1,71 @@
+package com.google.cloud.spark.spanner.binding;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Type;
+import com.google.cloud.spark.spanner.planning.expression.LiteralExpr;
+import java.math.BigDecimal;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.DecimalType;
+
+public final class SpannerTypeBinder {
+
+  public static void bind(Statement.Builder builder, String parameter, LiteralExpr literal) {
+
+    Object value = literal.getValue();
+    DataType type = literal.getSparkType();
+
+    if (value == null) {
+      builder.bind(parameter).to(mapToSpannerType(type), (com.google.cloud.spanner.Struct) null);
+      return;
+    }
+
+    if (type instanceof DecimalType) {
+      BigDecimal bd = (BigDecimal) value;
+      // Spanner NUMERIC validation: precision <= 38, scale <= 9
+      if (bd.precision() <= 38 && bd.scale() <= 9) {
+        builder.bind(parameter).to(bd);
+      } else {
+        throw new IllegalArgumentException("Decimal out of Spanner NUMERIC range");
+      }
+    } else if (type.sameType(DataTypes.StringType)) {
+      builder.bind(parameter).to((String) value);
+    } else if (type.sameType(DataTypes.LongType)) {
+      builder.bind(parameter).to((Long) value);
+    } else if (type.sameType(DataTypes.BooleanType)) {
+      builder.bind(parameter).to((Boolean) value);
+    } else if (type.sameType(DataTypes.IntegerType)) {
+      builder.bind(parameter).to(((Integer) value).longValue());
+    } else if (type.sameType(DataTypes.ShortType)) {
+      builder.bind(parameter).to(((Short) value).longValue());
+    } else if (type.sameType(DataTypes.ByteType)) {
+      builder.bind(parameter).to(((Byte) value).longValue());
+    } else if (type.sameType(DataTypes.DoubleType)) {
+      builder.bind(parameter).to((Double) value);
+    } else if (type.sameType(DataTypes.FloatType)) {
+      builder.bind(parameter).to(((Float) value).doubleValue());
+    } else {
+      throw new UnsupportedOperationException("Unsupported type: " + type);
+    }
+  }
+
+  private static Type mapToSpannerType(DataType type) {
+    if (type instanceof DecimalType) {
+      return Type.numeric();
+    } else if (type.sameType(DataTypes.StringType)) {
+      return Type.string();
+    } else if (type.sameType(DataTypes.LongType)
+        || type.sameType(DataTypes.IntegerType)
+        || type.sameType(DataTypes.ShortType)
+        || type.sameType(DataTypes.ByteType)) {
+      return Type.int64();
+    } else if (type.sameType(DataTypes.BooleanType)) {
+      return Type.bool();
+    } else if (type.sameType(DataTypes.DoubleType) || type.sameType(DataTypes.FloatType)) {
+      return Type.float64();
+    } else {
+      throw new UnsupportedOperationException(
+          "Unsupported Spark type for Spanner null mapping: " + type);
+    }
+  }
+}

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/planning/expression/ArithmeticExpr.java

@@ -26,16 +26,28 @@ public enum Operator {
     MOD
   }
 
-  Operator operator;
-  ValueExpr left;
-  ValueExpr right;
+  private Operator operator;
+  private ValueExpr left;
+  private ValueExpr right;
 
   public ArithmeticExpr(ValueExpr left, Operator operator, ValueExpr right) {
     this.left = Objects.requireNonNull(left, "left cannot be null");
     this.operator = Objects.requireNonNull(operator, "operator cannot be null");
     this.right = Objects.requireNonNull(right, "right cannot be null");
   }
 
+  public Operator getOperator() {
+    return operator;
+  }
+
+  public ValueExpr getLeft() {
+    return left;
+  }
+
+  public ValueExpr getRight() {
+    return right;
+  }
+
   @Override
   public <T> T accept(SpannerExprVisitor<T> visitor) {
     return visitor.visit(this);

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/planning/query/ExprConverterUtils.java

@@ -0,0 +1,28 @@
+package com.google.cloud.spark.spanner.planning.query;
+
+import com.google.cloud.spark.spanner.planning.expression.*;
+import org.apache.spark.sql.sources.*;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExprConverterUtils {
+  private static final Logger logger = LoggerFactory.getLogger(ExprConverterUtils.class);
+
+  public static ColumnExpr toColumn(String name, StructType schema) {
+    logger.debug("Looking up column '{}' in schema {}", name, schema.treeString());
+
+    StructField field = schema.apply(name);
+
+    return new ColumnExpr(name, field.dataType(), field.nullable());
+  }
+
+  public static LiteralExpr toLiteral(Object value, StructType schema, String columnName) {
+    logger.debug("Looking up literal column '{}' in schema {}", columnName, schema.treeString());
+
+    StructField field = schema.apply(columnName);
+
+    return new LiteralExpr(value, field.dataType());
+  }
+}

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/planning/query/LogicalQuery.java

@@ -14,15 +14,19 @@
 
 package com.google.cloud.spark.spanner.planning.query;
 
-import com.google.cloud.spark.spanner.scan.SpannerTable;
-import java.util.Map;
-import java.util.Set;
+import com.google.cloud.spark.spanner.planning.relation.Relation;
+import java.util.*;
 import org.apache.spark.sql.sources.Filter;
 import org.apache.spark.sql.types.StructField;
 
 public final class LogicalQuery {
-  public SpannerTable getSource() {
-    return spannerTable;
+  private final Relation source;
+  private final Set<String> requiredColumns;
+  private final Filter[] pushedFilters;
+  private final Map<String, StructField> fields;
+
+  public Relation getSource() {
+    return source;
   }
 
   public Set<String> getProjections() {
@@ -37,24 +41,54 @@ public Map<String, StructField> getFields() {
     return fields;
   }
 
-  private final SpannerTable spannerTable;
-  private final Set<String> requiredColumns;
-  private final Filter[] pushedFilters;
-  private final Map<String, StructField> fields;
+  private LogicalQuery(Builder builder) {
+    this.source = builder.source;
+    this.requiredColumns =
+        builder.requiredColumns != null
+            ? builder.requiredColumns
+            : java.util.Collections.emptySet();
+    this.pushedFilters =
+        builder.pushedFilters != null ? builder.pushedFilters.clone() : new Filter[0];
+    this.fields = builder.fields != null ? builder.fields : java.util.Collections.emptyMap();
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  public static final class Builder {
 
-  public LogicalQuery(
-      SpannerTable spannerTable,
-      Set<String> requiredColumns,
-      Filter[] pushedFilters,
-      Map<String, StructField> fields) {
+    private Relation source;
+    private Set<String> requiredColumns = Collections.emptySet();
+    private Filter[] pushedFilters = new Filter[0];
+    private Map<String, StructField> fields = java.util.Collections.emptyMap();
 
-    if (spannerTable == null) {
-      throw new NullPointerException("spannerTable cannot be null");
+    private Builder() {}
+
+    public Builder source(Relation source) {
+      this.source = source;
+      return this;
+    }
+
+    public Builder requiredColumns(Set<String> requiredColumns) {
+      this.requiredColumns = requiredColumns;
+      return this;
+    }
+
+    public Builder pushedFilters(Filter[] pushedFilters) {
+      this.pushedFilters = pushedFilters;
+      return this;
+    }
+
+    public Builder fields(Map<String, StructField> fields) {
+      this.fields = fields;
+      return this;
+    }
+
+    public LogicalQuery build() {
+      Objects.requireNonNull(source, "source");
+
+      return new LogicalQuery(this);
     }
-    this.spannerTable = spannerTable;
-    this.requiredColumns =
-        requiredColumns != null ? requiredColumns : java.util.Collections.emptySet();
-    this.pushedFilters = pushedFilters != null ? pushedFilters.clone() : new Filter[0];
-    this.fields = fields != null ? fields : java.util.Collections.emptyMap();
   }
 }

spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/rendering/GoogleSqlSpannerSqlExprVisitor.java

@@ -0,0 +1,40 @@
+package com.google.cloud.spark.spanner.rendering;
+
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spark.spanner.SpannerInformationSchema;
+import com.google.cloud.spark.spanner.binding.ParameterRef;
+import com.google.cloud.spark.spanner.binding.ParameterRegistry;
+import com.google.cloud.spark.spanner.planning.expression.LiteralExpr;
+import java.util.Collections;
+
+public class GoogleSqlSpannerSqlExprVisitor extends SqlExprVisitor {
+
+  public GoogleSqlSpannerSqlExprVisitor() {
+    super(
+        SpannerInformationSchema.create(Dialect.GOOGLE_STANDARD_SQL),
+        ParameterRegistry.create(Dialect.GOOGLE_STANDARD_SQL));
+  }
+
+  @Override
+  public RenderResult visit(LiteralExpr expr) {
+    ParameterRef ref = parameterRegistry.nextParameter();
+
+    return new RenderResult(
+        "@" + ref.getSqlName(), Collections.singletonMap(ref.getBindName(), expr));
+  }
+
+  @Override
+  public String renderStartsWith(String left, String right) {
+    return "STARTS_WITH(" + left + ", " + right + ")";
+  }
+
+  @Override
+  public String renderEndsWith(String left, String right) {
+    return "ENDS_WITH(" + left + ", " + right + ")";
+  }
+
+  @Override
+  public String renderContains(String left, String right) {
+    return "CONTAINS(" + left + ", " + right + ")";
+  }
+}