DRILL-8551: Fix Informix JDBC queries with catalog-qualified table names

contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDialectFactory.java

@@ -18,12 +18,16 @@
 package org.apache.drill.exec.store.jdbc;
 
 import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.dialect.InformixSqlDialect;
 import org.apache.drill.exec.store.jdbc.clickhouse.ClickhouseJdbcDialect;
+import org.apache.drill.exec.store.jdbc.informix.InformixJdbcDialect;
 
 import java.time.Duration;
+import java.util.Locale;
 
 public class JdbcDialectFactory {
   public static final String JDBC_CLICKHOUSE_PREFIX = "jdbc:clickhouse";
+  public static final String JDBC_INFORMIX_PREFIX = "jdbc:informix";
   public static final int CACHE_SIZE = 100;
   public static final Duration CACHE_TTL = Duration.ofHours(1);
   private volatile JdbcDialect jdbcDialect;
@@ -38,9 +42,16 @@ public JdbcDialect getJdbcDialect(JdbcStoragePlugin plugin, SqlDialect dialect)
       synchronized (this) {
         jd = jdbcDialect;
         if (jd == null) {
-          jd = plugin.getConfig().getUrl().startsWith(JDBC_CLICKHOUSE_PREFIX)
-                ? new ClickhouseJdbcDialect(plugin, dialect)
-                : new DefaultJdbcDialect(plugin, dialect);
+          String url = plugin.getConfig().getUrl();
+          String normalizedUrl = url == null ? null : url.toLowerCase(Locale.ROOT);
+          if (normalizedUrl != null && normalizedUrl.startsWith(JDBC_CLICKHOUSE_PREFIX)) {
+            jd = new ClickhouseJdbcDialect(plugin, dialect);
+          } else if (dialect instanceof InformixSqlDialect
+              || normalizedUrl != null && normalizedUrl.startsWith(JDBC_INFORMIX_PREFIX)) {
+            jd = new InformixJdbcDialect(plugin, dialect);
+          } else {
+            jd = new DefaultJdbcDialect(plugin, dialect);
+          }
           jdbcDialect = jd;
         }
       }

contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/informix/InformixJdbcDialect.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.jdbc.informix;
+
+import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.adapter.jdbc.JdbcImplementor;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.drill.exec.store.SubsetRemover;
+import org.apache.drill.exec.store.jdbc.DefaultJdbcDialect;
+import org.apache.drill.exec.store.jdbc.JdbcStoragePlugin;
+
+public class InformixJdbcDialect extends DefaultJdbcDialect {
+
+  private final SqlDialect dialect;
+
+  public InformixJdbcDialect(JdbcStoragePlugin plugin, SqlDialect dialect) {
+    super(plugin, dialect);
+    this.dialect = dialect;
+  }
+
+  @Override
+  public String generateSql(RelOptCluster cluster, RelNode input) {
+    final JdbcImplementor jdbcImplementor = new InformixJdbcImplementor(dialect,
+        (JavaTypeFactory) cluster.getTypeFactory());
+    final JdbcImplementor.Result result = jdbcImplementor.visitRoot(
+      input.accept(SubsetRemover.INSTANCE));
+    return result.asStatement().toSqlString(dialect).getSql();
+  }
+}

contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/informix/InformixJdbcImplementor.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.jdbc.informix;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.adapter.jdbc.JdbcImplementor;
+import org.apache.calcite.adapter.jdbc.JdbcTable;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+import com.google.common.collect.ImmutableList;
+
+import static java.util.Objects.requireNonNull;
+
+public class InformixJdbcImplementor extends JdbcImplementor {
+  public InformixJdbcImplementor(SqlDialect dialect, JavaTypeFactory typeFactory) {
+    super(dialect, typeFactory);
+  }
+
+  @Override
+  public Result visit(TableScan scan) {
+    SqlIdentifier sqlIdentifier = getSqlTargetTable(scan);
+    return result(sqlIdentifier, ImmutableList.of(Clause.FROM), scan, null);
+  }
+
+  static SqlIdentifier getSqlTargetTable(RelNode e) {
+    RelOptTable table = requireNonNull(e.getTable());
+    return table.maybeUnwrap(JdbcTable.class)
+      .map(jdbcTable -> {
+        List<String> names = new ArrayList<>(2);
+        // Informix JDBC reports the connected database as a catalog, but
+        // Informix SQL does not accept it as a dot-qualified table prefix.
+        if (jdbcTable.jdbcSchemaName != null) {
+          names.add(jdbcTable.jdbcSchemaName);
+        }
+        names.add(jdbcTable.jdbcTableName);
+        return new SqlIdentifier(names, SqlParserPos.ZERO);
+      })
+      .orElseGet(() -> new SqlIdentifier(table.getQualifiedName(), SqlParserPos.ZERO));
+  }
+}

contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/informix/TestInformixJdbcDialect.java

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.jdbc.informix;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Constructor;
+import java.util.Optional;
+
+import javax.sql.DataSource;
+
+import org.apache.calcite.adapter.jdbc.JdbcSchema;
+import org.apache.calcite.adapter.jdbc.JdbcTable;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.schema.Schema;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.dialect.InformixSqlDialect;
+import org.apache.drill.exec.store.jdbc.JdbcDialect;
+import org.apache.drill.exec.store.jdbc.JdbcDialectFactory;
+import org.apache.drill.exec.store.jdbc.JdbcStorageConfig;
+import org.apache.drill.exec.store.jdbc.JdbcStoragePlugin;
+import org.junit.Test;
+
+public class TestInformixJdbcDialect {
+
+  @Test
+  public void dropsCatalogFromTargetTable() throws Exception {
+    JdbcTable jdbcTable = newJdbcTable("sample_database", null, "sample_table");
+    TableScan scan = mock(TableScan.class);
+    RelOptTable relOptTable = mock(RelOptTable.class);
+    when(scan.getTable()).thenReturn(relOptTable);
+    when(relOptTable.maybeUnwrap(JdbcTable.class)).thenReturn(Optional.of(jdbcTable));
+
+    String sql = InformixJdbcImplementor.getSqlTargetTable(scan)
+      .toSqlString(InformixSqlDialect.DEFAULT)
+      .getSql();
+
+    assertEquals("sample_table", sql);
+  }
+
+  @Test
+  public void keepsSchemaWhenPresent() throws Exception {
+    JdbcTable jdbcTable = newJdbcTable("sample_database", "sample_owner", "sample_table");
+    TableScan scan = mock(TableScan.class);
+    RelOptTable relOptTable = mock(RelOptTable.class);
+    when(scan.getTable()).thenReturn(relOptTable);
+    when(relOptTable.maybeUnwrap(JdbcTable.class)).thenReturn(Optional.of(jdbcTable));
+
+    String sql = InformixJdbcImplementor.getSqlTargetTable(scan)
+      .toSqlString(InformixSqlDialect.DEFAULT)
+      .getSql();
+
+    assertEquals("sample_owner.sample_table", sql);
+  }
+
+  @Test
+  public void factorySelectsInformixDialect() {
+    JdbcStoragePlugin plugin = mock(JdbcStoragePlugin.class);
+    when(plugin.getConfig()).thenReturn(newJdbcStorageConfig("jdbc:other://localhost"));
+
+    JdbcDialect dialect = new JdbcDialectFactory().getJdbcDialect(plugin, InformixSqlDialect.DEFAULT);
+    assertTrue(dialect instanceof InformixJdbcDialect);
+  }
+
+  @Test
+  public void factorySelectsInformixDialectFromUrl() {
+    JdbcStoragePlugin plugin = mock(JdbcStoragePlugin.class);
+    when(plugin.getConfig()).thenReturn(
+      newJdbcStorageConfig("jdbc:informix-sqli://localhost:1526/sample_database"));
+
+    JdbcDialect dialect = new JdbcDialectFactory().getJdbcDialect(
+      plugin, SqlDialect.DatabaseProduct.UNKNOWN.getDialect());
+    assertTrue(dialect instanceof InformixJdbcDialect);
+  }
+
+  private static JdbcStorageConfig newJdbcStorageConfig(String url) {
+    JdbcStorageConfig config = new JdbcStorageConfig(
+      "com.informix.jdbc.IfxDriver",
+      url,
+      null,
+      null,
+      true,
+      false,
+      null,
+      null,
+      null,
+      0
+    );
+    return config;
+  }
+
+  private static JdbcTable newJdbcTable(String catalog, String schema, String table) throws Exception {
+    DataSource dataSource = mock(DataSource.class);
+    JdbcSchema jdbcSchema = new JdbcSchema(dataSource, InformixSqlDialect.DEFAULT, null, catalog, schema);
+    Constructor<JdbcTable> ctor = JdbcTable.class.getDeclaredConstructor(
+      JdbcSchema.class,
+      String.class,
+      String.class,
+      String.class,
+      Schema.TableType.class
+    );
+    ctor.setAccessible(true);
+    return ctor.newInstance(jdbcSchema, catalog, schema, table, Schema.TableType.TABLE);
+  }
+}