Skip to content

PHOENIX-6405 Disallow bulk loading into non-empty tables with global …#1265

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
richardantal merged 1 commit into from
Jul 19, 2021
Merged

PHOENIX-6405 Disallow bulk loading into non-empty tables with global … #1265

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 97 additions & 1 deletion phoenix-core/src/it/java/org/apache/phoenix/end2end/CsvBulkLoadToolIT.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -116,6 +116,101 @@ public void testBasicImport() throws Exception {
rs.close();
stmt.close();
}

@Test
public void testImportWithGlobalIndex() throws Exception {

Statement stmt = conn.createStatement();
stmt.execute("CREATE TABLE S.TABLE1 (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, T DATE) SPLIT ON (1,2)");
stmt.execute("CREATE INDEX glob_idx ON S.TABLE1(ID, T)");
conn.commit();

FileSystem fs = FileSystem.get(getUtility().getConfiguration());
FSDataOutputStream outputStream = fs.create(new Path("/tmp/input1.csv"));
PrintWriter printWriter = new PrintWriter(outputStream);
printWriter.println("1,Name 1,1970/01/01");
printWriter.println("2,Name 2,1970/01/02");
printWriter.close();

fs = FileSystem.get(getUtility().getConfiguration());
outputStream = fs.create(new Path("/tmp/input2.csv"));
printWriter = new PrintWriter(outputStream);
printWriter.println("3,Name 3,1970/01/03");
printWriter.println("4,Name 4,1970/01/04");
printWriter.close();

CsvBulkLoadTool csvBulkLoadTool = new CsvBulkLoadTool();
csvBulkLoadTool.setConf(new Configuration(getUtility().getConfiguration()));
csvBulkLoadTool.getConf().set(DATE_FORMAT_ATTRIB,"yyyy/MM/dd");
int exitCode = csvBulkLoadTool.run(new String[] {
"--input", "/tmp/input1.csv",
"--table", "table1",
"--schema", "s",
"--zookeeper", zkQuorum});
assertEquals(0, exitCode);

csvBulkLoadTool = new CsvBulkLoadTool();
csvBulkLoadTool.setConf(new Configuration(getUtility().getConfiguration()));
csvBulkLoadTool.getConf().set(DATE_FORMAT_ATTRIB,"yyyy/MM/dd");
try {
exitCode = csvBulkLoadTool.run(new String[] {
"--input", "/tmp/input2.csv",
"--table", "table1",
"--schema", "s",
"--zookeeper", zkQuorum});
fail("Bulk loading error should have happened earlier");
} catch (Exception e){
assertTrue(e.getMessage().contains("Bulk Loading error: Bulk loading is disabled for " +
"non empty tables with global indexes, because it will corrupt " +
"the global index table in most cases.\n" +
"Use the --corruptindexes option to override this check."));
}

ResultSet rs = stmt.executeQuery("SELECT id, name, t FROM s.table1 ORDER BY id");
assertTrue(rs.next());
assertEquals(1, rs.getInt(1));
assertEquals("Name 1", rs.getString(2));
assertEquals(DateUtil.parseDate("1970-01-01"), rs.getDate(3));
assertTrue(rs.next());
assertEquals(2, rs.getInt(1));
assertEquals("Name 2", rs.getString(2));
assertEquals(DateUtil.parseDate("1970-01-02"), rs.getDate(3));
assertFalse(rs.next());

csvBulkLoadTool = new CsvBulkLoadTool();
csvBulkLoadTool.setConf(new Configuration(getUtility().getConfiguration()));
csvBulkLoadTool.getConf().set(DATE_FORMAT_ATTRIB,"yyyy/MM/dd");
exitCode = csvBulkLoadTool.run(new String[] {
"--input", "/tmp/input2.csv",
"--table", "table1",
"--schema", "s",
"--zookeeper", zkQuorum,
"--corruptindexes"});
assertEquals(0, exitCode);

rs = stmt.executeQuery("SELECT id, name, t FROM s.table1 ORDER BY id");
assertTrue(rs.next());
assertEquals(1, rs.getInt(1));
assertEquals("Name 1", rs.getString(2));
assertEquals(DateUtil.parseDate("1970-01-01"), rs.getDate(3));
assertTrue(rs.next());
assertEquals(2, rs.getInt(1));
assertEquals("Name 2", rs.getString(2));
assertEquals(DateUtil.parseDate("1970-01-02"), rs.getDate(3));
assertTrue(rs.next());
assertEquals(3, rs.getInt(1));
assertEquals("Name 3", rs.getString(2));
assertEquals(DateUtil.parseDate("1970-01-03"), rs.getDate(3));
assertTrue(rs.next());
assertEquals(4, rs.getInt(1));
assertEquals("Name 4", rs.getString(2));
assertEquals(DateUtil.parseDate("1970-01-04"), rs.getDate(3));
assertFalse(rs.next());

rs.close();
stmt.close();
}

@Test
public void testImportWithRowTimestamp() throws Exception {

Expand DownExpand Up@@ -506,7 +601,8 @@ public void testImportWithDifferentPhysicalName() throws Exception {
exitCode = csvBulkLoadTool
.run(new String[] { "--input", csvNameForIndex, "--table", tableName,
"--schema", schemaName,
"--index-table", indexTableName, "--zookeeper", zkQuorum });
"--index-table", indexTableName, "--zookeeper", zkQuorum,
"--corruptindexes"});
assertEquals(0, exitCode);
selectFromIndex = "SELECT FIRST_NAME FROM " + fullTableName + " where FIRST_NAME='FirstName 3'";
rs = stmt.executeQuery("EXPLAIN " + selectFromIndex);
Expand Down
22 changes: 21 additions & 1 deletion phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -88,6 +88,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
static final Option IGNORE_ERRORS_OPT = new Option("g", "ignore-errors", false, "Ignore input errors");
static final Option HELP_OPT = new Option("h", "help", false, "Show this help and quit");
static final Option SKIP_HEADER_OPT = new Option("k", "skip-header", false, "Skip the first line of CSV files (the header)");
static final Option ENABLE_CORRUPT_INDEXES = new Option( "corruptindexes", "corruptindexes", false, "Allow bulk loading into non-empty tables with global secondary indexes");

/**
* Set configuration values based on parsed command line options.
Expand All@@ -112,6 +113,7 @@ protected Options getOptions() {
options.addOption(IGNORE_ERRORS_OPT);
options.addOption(HELP_OPT);
options.addOption(SKIP_HEADER_OPT);
options.addOption(ENABLE_CORRUPT_INDEXES);
return options;
}

Expand DownExpand Up@@ -226,6 +228,12 @@ private int loadData(Configuration conf, CommandLine cmdLine) throws Exception {
configureOptions(cmdLine, importColumns, conf);
String sName = SchemaUtil.normalizeIdentifier(schemaName);
String tName = SchemaUtil.normalizeIdentifier(tableName);

String tn = SchemaUtil.getEscapedTableName(sName, tName);
ResultSet rsempty = conn.createStatement().executeQuery("SELECT * FROM " + tn + " LIMIT 1");
boolean tableNotEmpty = rsempty.next();
rsempty.close();

try {
validateTable(conn, sName, tName);
} finally {
Expand All@@ -244,14 +252,26 @@ private int loadData(Configuration conf, CommandLine cmdLine) throws Exception {
PTable table = PhoenixRuntime.getTable(conn, qualifiedTableName);
tablesToBeLoaded.add(new TargetTableRef(qualifiedTableName, table.getPhysicalName().getString()));
boolean hasLocalIndexes = false;
boolean hasGlobalIndexes = false;
for(PTable index: table.getIndexes()) {
if (index.getIndexType() == IndexType.LOCAL) {
hasLocalIndexes =
qualifiedIndexTableName == null ? true : index.getTableName().getString()
.equals(qualifiedIndexTableName);
if (hasLocalIndexes) break;
if (hasLocalIndexes && hasGlobalIndexes) break;
}
if (index.getIndexType() == IndexType.GLOBAL) {
hasGlobalIndexes = true;
if (hasLocalIndexes && hasGlobalIndexes) break;
}
}

if(hasGlobalIndexes && tableNotEmpty && !cmdLine.hasOption(ENABLE_CORRUPT_INDEXES.getOpt())){
throw new IllegalStateException("Bulk Loading error: Bulk loading is disabled for non" +
" empty tables with global indexes, because it will corrupt the global index table in most cases.\n" +
"Use the --corruptindexes option to override this check.");
}

// using conn after it's been closed... o.O
tablesToBeLoaded.addAll(getIndexTables(conn, qualifiedTableName));

Expand Down
2 changes: 1 addition & 1 deletion phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -673,7 +673,7 @@ public static String getEscapedTableName(String schemaName, String tableName) {
if (schemaName == null || schemaName.length() == 0) {
return "\"" + tableName + "\"";
}
return "\"" + schemaName + "\"." + "\"" + tableName + "\"";
return "\"" + schemaName + "\"" + QueryConstants.NAME_SEPARATOR + "\"" + tableName + "\"";
}

protected static PhoenixConnection addMetaDataColumn(PhoenixConnection conn, long scn, String columnDef) throws SQLException {
Expand Down