|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | + * or more contributor license agreements. See the NOTICE file |
| 4 | + * distributed with this work for additional information |
| 5 | + * regarding copyright ownership. The ASF licenses this file |
| 6 | + * to you under the Apache License, Version 2.0 (the |
| 7 | + * "License"); you may not use this file except in compliance |
| 8 | + * with the License. You may obtain a copy of the License at |
| 9 | + * |
| 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | + * |
| 12 | + * Unless required by applicable law or agreed to in writing, software |
| 13 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 14 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 15 | + * See the License for the specific language governing permissions and |
| 16 | + * limitations under the License. |
| 17 | + */ |
| 18 | + |
| 19 | +package org.apache.seatunnel.connectors.seatunnel.jdbc; |
| 20 | + |
| 21 | +import org.apache.seatunnel.api.table.type.SeaTunnelRow; |
| 22 | +import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleCatalog; |
| 23 | +import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleURLParser; |
| 24 | + |
| 25 | +import org.apache.commons.lang3.tuple.Pair; |
| 26 | + |
| 27 | +import org.testcontainers.containers.GenericContainer; |
| 28 | +import org.testcontainers.containers.OracleContainer; |
| 29 | +import org.testcontainers.containers.output.Slf4jLogConsumer; |
| 30 | +import org.testcontainers.utility.DockerImageName; |
| 31 | +import org.testcontainers.utility.DockerLoggerFactory; |
| 32 | +import org.testcontainers.utility.MountableFile; |
| 33 | + |
| 34 | +import com.google.common.collect.Lists; |
| 35 | + |
| 36 | +import java.math.BigDecimal; |
| 37 | +import java.sql.Date; |
| 38 | +import java.sql.Timestamp; |
| 39 | +import java.time.LocalDate; |
| 40 | +import java.time.LocalDateTime; |
| 41 | +import java.util.ArrayList; |
| 42 | +import java.util.HashMap; |
| 43 | +import java.util.List; |
| 44 | +import java.util.Map; |
| 45 | + |
| 46 | +public class JdbcOracleLowercaseTableIT extends AbstractJdbcIT { |
| 47 | + |
| 48 | + private static final String ORACLE_IMAGE = "gvenzl/oracle-xe:21-slim-faststart"; |
| 49 | + private static final String ORACLE_NETWORK_ALIASES = "e2e_oracleDb"; |
| 50 | + private static final String DRIVER_CLASS = "oracle.jdbc.OracleDriver"; |
| 51 | + private static final int ORACLE_PORT = 1521; |
| 52 | + private static final String ORACLE_URL = "jdbc:oracle:thin:@" + HOST + ":%s/%s"; |
| 53 | + private static final String USERNAME = "TESTUSER"; |
| 54 | + private static final String PASSWORD = "testPassword"; |
| 55 | + private static final String DATABASE = "XE"; |
| 56 | + private static final String SCHEMA = USERNAME; |
| 57 | + private static final String SOURCE_TABLE = "E2E_TABLE_SOURCE_LOWER"; |
| 58 | + private static final String SINK_TABLE = "E2E_TABLE_SINK_LOWER"; |
| 59 | + private static final String CATALOG_TABLE = "e2e_table_catalog_lower"; |
| 60 | + // no execute conf just test lower oracle create table |
| 61 | + private static final List<String> CONFIG_FILE = Lists.newArrayList(); |
| 62 | + |
| 63 | + private static final String CREATE_SQL = |
| 64 | + "create table %s\n" |
| 65 | + + "(\n" |
| 66 | + + " VARCHAR_10_COL varchar2(10),\n" |
| 67 | + + " CHAR_10_COL char(10),\n" |
| 68 | + + " CLOB_COL clob,\n" |
| 69 | + + " NUMBER_3_SF_2_DP number(3, 2),\n" |
| 70 | + + " INTEGER_COL integer,\n" |
| 71 | + + " FLOAT_COL float(10),\n" |
| 72 | + + " REAL_COL real,\n" |
| 73 | + + " BINARY_FLOAT_COL binary_float,\n" |
| 74 | + + " BINARY_DOUBLE_COL binary_double,\n" |
| 75 | + + " DATE_COL date,\n" |
| 76 | + + " TIMESTAMP_WITH_3_FRAC_SEC_COL timestamp(3),\n" |
| 77 | + + " TIMESTAMP_WITH_LOCAL_TZ timestamp with local time zone\n" |
| 78 | + + ")"; |
| 79 | + |
| 80 | + @Override |
| 81 | + JdbcCase getJdbcCase() { |
| 82 | + Map<String, String> containerEnv = new HashMap<>(); |
| 83 | + containerEnv.put("ORACLE_PASSWORD", PASSWORD); |
| 84 | + containerEnv.put("APP_USER", USERNAME); |
| 85 | + containerEnv.put("APP_USER_PASSWORD", PASSWORD); |
| 86 | + String jdbcUrl = String.format(ORACLE_URL, ORACLE_PORT, SCHEMA); |
| 87 | + Pair<String[], List<SeaTunnelRow>> testDataSet = initTestData(); |
| 88 | + String[] fieldNames = testDataSet.getKey(); |
| 89 | + |
| 90 | + String insertSql = insertTable(SCHEMA, SOURCE_TABLE, fieldNames); |
| 91 | + |
| 92 | + return JdbcCase.builder() |
| 93 | + .dockerImage(ORACLE_IMAGE) |
| 94 | + .networkAliases(ORACLE_NETWORK_ALIASES) |
| 95 | + .containerEnv(containerEnv) |
| 96 | + .driverClass(DRIVER_CLASS) |
| 97 | + .host(HOST) |
| 98 | + .port(ORACLE_PORT) |
| 99 | + .localPort(ORACLE_PORT) |
| 100 | + .jdbcTemplate(ORACLE_URL) |
| 101 | + .jdbcUrl(jdbcUrl) |
| 102 | + .userName(USERNAME) |
| 103 | + .password(PASSWORD) |
| 104 | + .database(DATABASE) |
| 105 | + .schema(SCHEMA) |
| 106 | + .sourceTable(SOURCE_TABLE) |
| 107 | + .sinkTable(SINK_TABLE) |
| 108 | + .catalogDatabase(DATABASE) |
| 109 | + .catalogSchema(SCHEMA) |
| 110 | + .catalogTable(CATALOG_TABLE) |
| 111 | + .createSql(CREATE_SQL) |
| 112 | + .configFile(CONFIG_FILE) |
| 113 | + .insertSql(insertSql) |
| 114 | + .testData(testDataSet) |
| 115 | + .build(); |
| 116 | + } |
| 117 | + |
| 118 | + @Override |
| 119 | + void compareResult() {} |
| 120 | + |
| 121 | + @Override |
| 122 | + String driverUrl() { |
| 123 | + return "https://repo1.maven.org/maven2/com/oracle/database/jdbc/ojdbc8/12.2.0.1/ojdbc8-12.2.0.1.jar"; |
| 124 | + } |
| 125 | + |
| 126 | + @Override |
| 127 | + Pair<String[], List<SeaTunnelRow>> initTestData() { |
| 128 | + String[] fieldNames = |
| 129 | + new String[] { |
| 130 | + "VARCHAR_10_COL", |
| 131 | + "CHAR_10_COL", |
| 132 | + "CLOB_COL", |
| 133 | + "NUMBER_3_SF_2_DP", |
| 134 | + "INTEGER_COL", |
| 135 | + "FLOAT_COL", |
| 136 | + "REAL_COL", |
| 137 | + "BINARY_FLOAT_COL", |
| 138 | + "BINARY_DOUBLE_COL", |
| 139 | + "DATE_COL", |
| 140 | + "TIMESTAMP_WITH_3_FRAC_SEC_COL", |
| 141 | + "TIMESTAMP_WITH_LOCAL_TZ" |
| 142 | + }; |
| 143 | + |
| 144 | + List<SeaTunnelRow> rows = new ArrayList<>(); |
| 145 | + for (int i = 0; i < 100; i++) { |
| 146 | + SeaTunnelRow row = |
| 147 | + new SeaTunnelRow( |
| 148 | + new Object[] { |
| 149 | + String.format("f%s", i), |
| 150 | + String.format("f%s", i), |
| 151 | + String.format("f%s", i), |
| 152 | + BigDecimal.valueOf(1.1), |
| 153 | + i, |
| 154 | + Float.parseFloat("2.2"), |
| 155 | + Float.parseFloat("2.2"), |
| 156 | + Float.parseFloat("22.2"), |
| 157 | + Double.parseDouble("2.2"), |
| 158 | + Date.valueOf(LocalDate.now()), |
| 159 | + Timestamp.valueOf(LocalDateTime.now()), |
| 160 | + Timestamp.valueOf(LocalDateTime.now()) |
| 161 | + }); |
| 162 | + rows.add(row); |
| 163 | + } |
| 164 | + |
| 165 | + return Pair.of(fieldNames, rows); |
| 166 | + } |
| 167 | + |
| 168 | + @Override |
| 169 | + GenericContainer<?> initContainer() { |
| 170 | + DockerImageName imageName = DockerImageName.parse(ORACLE_IMAGE); |
| 171 | + |
| 172 | + GenericContainer<?> container = |
| 173 | + new OracleContainer(imageName) |
| 174 | + .withDatabaseName(SCHEMA) |
| 175 | + .withCopyFileToContainer( |
| 176 | + MountableFile.forClasspathResource("sql/oracle_init.sql"), |
| 177 | + "/container-entrypoint-startdb.d/init.sql") |
| 178 | + .withNetwork(NETWORK) |
| 179 | + .withNetworkAliases(ORACLE_NETWORK_ALIASES) |
| 180 | + .withExposedPorts(ORACLE_PORT) |
| 181 | + .withLogConsumer( |
| 182 | + new Slf4jLogConsumer(DockerLoggerFactory.getLogger(ORACLE_IMAGE))); |
| 183 | + |
| 184 | + container.setPortBindings( |
| 185 | + Lists.newArrayList(String.format("%s:%s", ORACLE_PORT, ORACLE_PORT))); |
| 186 | + |
| 187 | + return container; |
| 188 | + } |
| 189 | + |
| 190 | + @Override |
| 191 | + public String quoteIdentifier(String field) { |
| 192 | + return "\"" + field + "\""; |
| 193 | + } |
| 194 | + |
| 195 | + @Override |
| 196 | + protected void clearTable(String database, String schema, String table) { |
| 197 | + clearTable(schema, table); |
| 198 | + } |
| 199 | + |
| 200 | + @Override |
| 201 | + protected String buildTableInfoWithSchema(String database, String schema, String table) { |
| 202 | + return buildTableInfoWithSchema(schema, table); |
| 203 | + } |
| 204 | + |
| 205 | + @Override |
| 206 | + protected void initCatalog() { |
| 207 | + String jdbcUrl = jdbcCase.getJdbcUrl().replace(HOST, dbServer.getHost()); |
| 208 | + catalog = |
| 209 | + new OracleCatalog( |
| 210 | + "oracle", |
| 211 | + jdbcCase.getUserName(), |
| 212 | + jdbcCase.getPassword(), |
| 213 | + OracleURLParser.parse(jdbcUrl), |
| 214 | + SCHEMA); |
| 215 | + catalog.open(); |
| 216 | + } |
| 217 | +} |
0 commit comments