arina-ielchiieva commented on a change in pull request #1711: DRILL-7011: Support schema in scan framework URL: https://github.com/apache/drill/pull/1711#discussion_r268604275
########## File path: exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java ########## @@ -82,6 +167,468 @@ public void testSchema() throws Exception { .addRow(10, new LocalDate(2019, 3, 20), "it works!", 1234.5D, 20L, "") .build(); RowSetUtilities.verify(expected, actual); + } finally { + resetV3(); + resetSchema(); + } + } + + + /** + * Use a schema with explicit projection to get a consistent view + * of the table schema, even if columns are missing, rows are ragged, + * and column order changes. + * <p> + * Force the scans to occur in distinct fragments so the order of the + * file batches is random. + */ + @Test + public void testMultiFileSchema() throws Exception { + RowSet expected1 = null; + RowSet expected2 = null; + try { + enableV3(true); + enableSchema(true); + enableMultiScan(); + String tablePath = buildTwoFileTable("multiFileSchema", raggedMulti1Contents, reordered2Contents); + run(SCHEMA_SQL, tablePath); + + // Wildcard expands to union of schema + table. In this case + // all table columns appear in the schema (though not all schema + // columns appear in the table.) + + String sql = "SELECT id, `name`, `date`, gender, comment FROM " + tablePath; + TupleMetadata expectedSchema = new SchemaBuilder() + .add("id", MinorType.INT) + .add("name", MinorType.VARCHAR) + .addNullable("date", MinorType.DATE) + .add("gender", MinorType.VARCHAR) + .add("comment", MinorType.VARCHAR) + .buildSchema(); + expected1 = new RowSetBuilder(client.allocator(), expectedSchema) + .addRow(1, "arina", new LocalDate(2019, 1, 18), "female", "ABC") + .addRow(2, "javan", new LocalDate(2019, 1, 19), "male", "ABC") + .addRow(4, "albert", new LocalDate(2019, 5, 4), "", "ABC") + .build(); + expected2 = new RowSetBuilder(client.allocator(), expectedSchema) + .addRow(3, "bob", new LocalDate(2001, 1, 16), "NA", "ABC") + .build(); + + // Loop 10 times so that, as the two reader fragments read the two + // files, we end up with (acceptable) races that read the files in + // random order. + + for (int i = 0; i < 10; i++) { + boolean sawSchema = false; + boolean sawFile1 = false; + boolean sawFile2 = false; + Iterator<DirectRowSet> iter = client.queryBuilder().sql(sql).rowSetIterator(); + while (iter.hasNext()) { + RowSet result = iter.next(); + if (result.rowCount() == 3) { + sawFile1 = true; + new RowSetComparison(expected1).verifyAndClear(result); + } else if (result.rowCount() == 1) { + sawFile2 = true; + new RowSetComparison(expected2).verifyAndClear(result); + } else { + assertEquals(0, result.rowCount()); + sawSchema = true; + } + } + assertTrue(sawSchema); + assertTrue(sawFile1); + assertTrue(sawFile2); + } + } finally { + expected1.clear(); + expected2.clear(); + client.resetSession(ExecConstants.ENABLE_V3_TEXT_READER_KEY); + client.resetSession(ExecConstants.STORE_TABLE_USE_SCHEMA_FILE); + client.resetSession(ExecConstants.MIN_READER_WIDTH_KEY); + } + } + + /** + * Test the schema we get in V2 when the table read order is random. + * Worst-case: the two files have different column counts and + * column orders. + * <p> + * Though the results are random, we iterate 10 times which, in most runs, + * shows the random variation in schemas: + * <ul> + * <li>Sometimes the first batch has three columns, sometimes four.</li> + * <li>Sometimes the column `id` is in position 0, sometimes in position 1 + * (correlated with the above).</li> + * <li>Due to the fact that sometimes the first file (with four columns) + * is returned first, sometimes the second file (with three columns) is + * returned first.</li> + * </ul> + */ + @Test + public void testSchemaRaceV2() throws Exception { + try { + enableV3(false); + enableSchema(false); + enableMultiScan(); + String tablePath = buildTwoFileTable("schemaRaceV2", multi1Contents, reordered2Contents); + boolean sawFile1First = false; + boolean sawFile2First = false; + boolean sawFullSchema = false; + boolean sawPartialSchema = false; + boolean sawIdAsCol0 = false; + boolean sawIdAsCol1 = false; + String sql = "SELECT * FROM " + tablePath; + for (int i = 0; i < 10; i++) { + Iterator<DirectRowSet> iter = client.queryBuilder().sql(sql).rowSetIterator(); + int batchCount = 0; + while(iter.hasNext()) { + batchCount++; + RowSet result = iter.next(); + TupleMetadata resultSchema = result.schema(); + if (resultSchema.size() == 4) { + sawFullSchema = true; + } else { + assertEquals(3, resultSchema.size()); + sawPartialSchema = true; + } + if (resultSchema.index("id") == 0) { + sawIdAsCol0 = true; + } else { + assertEquals(1, resultSchema.index("id")); + sawIdAsCol1 = true; + } + if (batchCount == 1) { + RowSetReader reader = result.reader(); + assertTrue(reader.next()); + String id = reader.scalar("id").getString(); + if (id.equals("1")) { + sawFile1First = true; + } else { + assertEquals("3", id); + sawFile2First = true; + } + } + result.clear(); + } + } + + // Outcome is random (which is the key problem). Don't assert on these + // because doing so can lead to a flakey test. + + if (!sawFile1First || ! sawFile2First || !sawFullSchema || !sawPartialSchema || !sawIdAsCol0 || !sawIdAsCol1) { + System.out.println("Some variations did not occur"); + System.out.println(String.format("File 1 first: %s", sawFile1First)); + System.out.println(String.format("File 1 second: %s", sawFile2First)); + System.out.println(String.format("Full schema: %s", sawFullSchema)); + System.out.println(String.format("Partial schema: %s", sawPartialSchema)); + System.out.println(String.format("`id` as col 0: %s", sawIdAsCol0)); + System.out.println(String.format("`id` as col 1: %s", sawIdAsCol1)); + } + // Sanity checks + assertTrue(sawFullSchema); + assertTrue(sawFile1First || sawFile2First); + assertTrue(sawIdAsCol0 || sawIdAsCol1); + } finally { + resetV3(); + resetSchema(); + resetMultiScan(); + } + } + + /** + * Show that, without schema, the hard schema change for the "missing" + * gender column causes an error in the sort operator when presented with + * one batch in which gender is VARCHAR, another in which it is + * Nullable INT. This is a consequence of using SELECT * on a distributed + * scan. + */ + @Test + public void testWildcardSortFailure() throws Exception { + try { + enableSchema(false); + enableMultiScan(); + enableV3(false); + String tablePath = buildTwoFileTable("wildcardSortV2", multi1Contents, reordered2Contents); + doTestWildcardSortFailure(tablePath); + enableV3(true); + doTestWildcardSortFailure(tablePath); + } finally { + resetV3(); + resetSchema(); + resetMultiScan(); + } + } + + private void doTestWildcardSortFailure(String tablePath) throws Exception { + String sql = "SELECT * FROM " + tablePath + " ORDER BY id"; + boolean sawError = false; + for (int i = 0; i < 10; i++) { + try { + // When this fails it will print a nasty stack trace. + RowSet result = client.queryBuilder().sql(sql).rowSet(); + assertEquals(4, result.rowCount()); + result.clear(); + } catch (RpcException e) { + assertTrue(e.getCause() instanceof UserRemoteException); + sawError = true; + break; + } + } + assertTrue(sawError); + } + + /** + * Test an explicit projection with a sort. Using the sort 1) will blow up + * if the internal schema is inconsistent, and 2) allows easier verification + * of the merged table results. + * <p> + * Fails with <code><pre> + * #: id, name, gender + * 0: "1", "bob