Fix skipping in diverse scalar stream readers #12721
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
- Extracted pushdown subscripts logic into its own rule, PushDownSubscripts, from PruneUnreferencedOutputs. - Replaced Subfield.PathElement with a hierarchy of classes: empty base class PathElement + NestedField, IntegerSubscript, StringSubscript sub-classes. NestedField represents subfield of a struct: c.a, IntegerSubscript - element of an array or map with integer keys: c[2], StringSubscript - element of a map with varchar keys c["a"].
Rename PushdownSubscripts rule into PushdownSubfields. Change its logic to create project node on top of scan using new filter_by_subfield_paths function instead of modifying column handles in table scan node via calls to ColumnHandle.createSubfieldPruningColumnHandle. Remove ColumnHandle.createSubfieldPruningColumnHandle API.
Add an optimizer rule to pushdown subfield pruning into connectors. The new rule uses the new metadata API pushdownSubfieldPruning.
OrcRecordReader.getNextPage tracks the average bytes per rows of initial qualifying set. The batch size is adapted to stay within expected soft limits. If a hard limit is exceeded an exception is thrown and the streams are reset to the beginning of the current row group. This behavior only works when the streams are checkpointed, hence ValueStreamSources are not used even if the stripe consisted of a single row group. This replaces the StreamReader result truncation mechanism.
mbasmanova
changed the title
|
@oerling I'm seeing a test failure: Does it work locally? I'm going to re-start the build in case this error is transient. |
|
@elonazoulay Elon, would you take a look at this change? I feel that we should be able to build a unit test that catches this type of error. |
|
This is timing related flakiness. Not locally reproducible and no relation to the diff.
If this comes from some shared infra this will likely never reproduce on a local setup because this would depend on some unusual timing caused by system load.
From: Maria Basmanova <notifications@github.com>
Sent: Wednesday, April 24, 2019 10:34 AM
To: prestodb/presto <presto@noreply.github.com>
Cc: oerling <erling@xs4all.nl>; Mention <mention@noreply.github.com>
Subject: Re: [prestodb/presto] Fix skipping in diverse scalar stream readers (#12721)
@oerling <https://github.com/oerling> I'm seeing a test failure:
[ERROR] Tests run: 5662, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 1,255.582 s <<< FAILURE! - in TestSuite
[ERROR] testRemoveRemoteSource(com.facebook.presto.operator.TestExchangeClient) Time elapsed: 0.012 s <<< FAILURE!
java.lang.AssertionError: expected [4] but found [2]
at org.testng.Assert.fail(Assert.java:94)
at org.testng.Assert.failNotEquals(Assert.java:513)
at org.testng.Assert.assertEqualsImpl(Assert.java:135)
at org.testng.Assert.assertEquals(Assert.java:116)
at org.testng.Assert.assertEquals(Assert.java:389)
at org.testng.Assert.assertEquals(Assert.java:399)
at com.facebook.presto.operator.TestExchangeClient.assertPageEquals(TestExchangeClient.java:416)
at com.facebook.presto.operator.TestExchangeClient.testRemoveRemoteSource(TestExchangeClient.java:379)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:104)
at org.testng.internal.Invoker.invokeMethod(Invoker.java:645)
at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:851)
at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1177)
at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:129)
at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:112)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Does it work locally? I'm going to re-start the build in case this error is transient.
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub <#12721 (comment)> , or mute the thread <https://github.com/notifications/unsubscribe-auth/AKPPPT5CPV5OVGNAILLOASLPSCKZBANCNFSM4HIAA2VQ> .
|
1. If results are returned from a Slice dictionary column, there must be a Page break if the underlying dictionary changes. 2. The in-dictionary flags were read twice and in a sequence that does not correspond to the input qualifying set. Add consistent resize and initial size setting for reused and dynamically growing arrays.
Indeed. Looks like so. The test passed on retry. |
Moves common functions between direct and dictionary long stream readers to a common superclass.
mbasmanova
requested changes
Apr 26, 2019
There was a problem hiding this comment.
Adds a test. Note that the test is not supported by H2 and requires comparing Presto with and without Aria scan.
Aria scan cannot have a dependency on default (soon to be legacy) scan. Hence, we need to write tests some other way.
For this particular case, a unit test would be most appropriate, although H2 works just fine as well:
private static final String LINEITEM_ARIA_H2_CTE = "WITH lineitem_aria AS (SELECT\n" +
" orderkey,\n" +
" partkey,\n" +
" suppkey,\n" +
" linenumber,\n" +
" quantity,\n" +
" extendedprice,\n" +
" shipmode,\n" +
" shipdate,\n" +
" comment,\n" +
" returnflag = 'R' AS is_returned,\n" +
" CAST(quantity + 1 AS REAL) AS float_quantity,\n" +
" CAST(discount AS decimal(5, 2)) AS short_decimal_discount,\n" +
" CAST(discount AS decimal(38, 2)) AS long_decimal_discount,\n" +
" CAST(CASEWHEN(shipmode = 'SHIP', 1, CASEWHEN(shipmode = 'REG AIR', 2, CASEWHEN(shipmode = 'AIR', 3, CASEWHEN(shipmode = 'FOB', 4, CASEWHEN(shipmode = 'MAIL', 5, CASEWHEN(shipmode = 'RAIL', 6, CASEWHEN(shipmode = 'TRUCK', 7, null))))))) AS tinyint) AS tinyint_shipmode,\n" +
" dateadd(SECOND, suppkey, cast(shipdate as timestamp)) AS timestamp_shipdate\n" +
"FROM lineitem)\n";
String query = "SELECT partkey, shipmode, comment, long_decimal_discount, short_decimal_discount, " +
"is_returned, float_quantity, timestamp_shipdate, tinyint_shipmode " +
"FROM lineitem_aria WHERE linenumber = 4";
assertQuery(ariaSession(), query, LINEITEM_ARIA_H2_CTE + query);
The above test reproduces the issue as expected:
Caused by: com.facebook.presto.orc.OrcCorruptionException: Malformed ORC file. Reading BigInteger past EOF [file:/Users/mbasmanova/aria/test_data3/hive_data/tpch/lineitem_aria/20190314_055037_00008_zuqv6_afaded0a-76a8-435e-a41c-dd582fa33ca3]
at com.facebook.presto.orc.stream.DecimalInputStream.skip(DecimalInputStream.java:165)
at com.facebook.presto.orc.reader.DecimalStreamReader.scan(DecimalStreamReader.java:309)
at com.facebook.presto.orc.ColumnGroupReader.advance(ColumnGroupReader.java:413)
at com.facebook.presto.orc.OrcRecordReader.getNextPage(OrcRecordReader.java:828)
at com.facebook.presto.hive.orc.OrcPageSource.getNextPage(OrcPageSource.java:150)
... 13 more
For queries that need to access complex types (arrays, maps, structs), we could compare Hive with Tpch, e.g.
private static final String LINEITEM_ARIA_SQL = "SELECT\n" +
" orderkey,\n" +
" partkey,\n" +
" suppkey,\n" +
" linenumber,\n" +
" quantity,\n" +
" extendedprice,\n" +
" shipmode,\n" +
" shipdate,\n" +
" comment,\n" +
" returnflag = 'R' AS is_returned,\n" +
" CAST(quantity + 1 AS REAL) AS float_quantity,\n" +
" CAST(discount AS decimal(5, 2)) AS short_decimal_discount,\n" +
" CAST(discount AS decimal(38, 2)) AS long_decimal_discount,\n" +
" CAST(array_position(array['SHIP','REG AIR','AIR','FOB','MAIL','RAIL','TRUCK'], shipmode) AS tinyint) AS tinyint_shipmode,\n" +
" date_add('second', suppkey, cast(shipdate as timestamp)) AS timestamp_shipdate,\n" +
" MAP(\n" +
" ARRAY[1, 2, 3],\n" +
" ARRAY[orderkey, partkey, suppkey]\n" +
" ) AS order_part_supp_map,\n" +
" ARRAY[ARRAY[orderkey, partkey, suppkey]] AS order_part_supp_array\n" +
"FROM tpch.tiny.lineitem";
private static final String LINEITEM_ARIA_CTE = format("WITH lineitem_aria AS (%s)\n", LINEITEM_ARIA_SQL);
Session tpchSession = Session.builder(getQueryRunner().getDefaultSession())
.setCatalog("tpch")
.setSchema("tiny")
.build();
assertQuery(ariaSession(), query, tpchSession, LINEITEM_ARIA_CTE + query, "SELECT 1");
|
Well, since the data comes from TPC H anyway one can compare what was read to what was written.
I just want a formulation that does not involve 50 line SQL statements as string literals.
Everybody else has done this sort of thing with literal queries and reference outputs In a file.
So here it is possible to have templates that get variously substituted, e.g. table names and filters and derived tables. Preferrably these come from a file. The datasets are, per the table, a base query against TPC H that can be a join. Look at making cust_order_line in scripts/ aria-research. Then there is a stack of derived tables enclosing this. Each adds select and optional group by.
On one side you have the original. On the other side you have the select from the result insert – original. These have a name, e.g. cust_line_order.
You write
select count (*) from cust_line_order cross join unnest (orders) where orders.o_date between xx and yy.
This is the whole definition of the test. This gets evaluated against the table and against the derived table that made the table. This can work work with miniml scale.
Then you declare cust_line_order as something like
Cust_line_order
Select c_custkey, c_acctbal, array_aagg(row(o_orderkey, o_orderdate,…))) as orders from (
Select … from customer, orders where o_custkey = o_custkey
Group by c_custkey, …)
The test framework can then decide what to do. In the simplest case it can run the query with the generating DYT in the place of the table to get the reference output. It can also store this reference output in some place that is considered stable. For testing Aria ORC, Parquet would be a candidate.
And then your test becomes one file with the dataset definitions and another with the queries.
Now this is easy to write and uses the assert… functions and does not have SQL as string literals.
Also note that we need datasets that have weird distribbutions of lengths. These must be a certain size. These can all come from TPC H. But regenerating these for every query for making the reference dataset is impractical, so the framework will have to actually store these.
Furthermore then tests can be run with a single command and do not involve compiling code.
From: Maria Basmanova <notifications@github.com>
Sent: Friday, April 26, 2019 8:34 AM
To: prestodb/presto <presto@noreply.github.com>
Cc: oerling <erling@xs4all.nl>; Mention <mention@noreply.github.com>
Subject: Re: [prestodb/presto] Fix skipping in diverse scalar stream readers (#12721)
@mbasmanova requested changes on this pull request.
@oerling <https://github.com/oerling>
Adds a test. Note that the test is not supported by H2 and requires comparing Presto with and without Aria scan.
Aria scan cannot have a dependency on default (soon to be legacy) scan. Hence, we need to write tests some other way.
For this particular case, a unit test would be most appropriate, although H2 works just fine as well:
private static final String LINEITEM_ARIA_H2_CTE = "WITH lineitem_aria AS (SELECT\n" +
" orderkey,\n" +
" partkey,\n" +
" suppkey,\n" +
" linenumber,\n" +
" quantity,\n" +
" extendedprice,\n" +
" shipmode,\n" +
" shipdate,\n" +
" comment,\n" +
" returnflag = 'R' AS is_returned,\n" +
" CAST(quantity + 1 AS REAL) AS float_quantity,\n" +
" CAST(discount AS decimal(5, 2)) AS short_decimal_discount,\n" +
" CAST(discount AS decimal(38, 2)) AS long_decimal_discount,\n" +
" CAST(CASEWHEN(shipmode = 'SHIP', 1, CASEWHEN(shipmode = 'REG AIR', 2, CASEWHEN(shipmode = 'AIR', 3, CASEWHEN(shipmode = 'FOB', 4, CASEWHEN(shipmode = 'MAIL', 5, CASEWHEN(shipmode = 'RAIL', 6, CASEWHEN(shipmode = 'TRUCK', 7, null))))))) AS tinyint) AS tinyint_shipmode,\n" +
" dateadd(SECOND, suppkey, cast(shipdate as timestamp)) AS timestamp_shipdate\n" +
"FROM lineitem)\n";
String query = "SELECT partkey, shipmode, comment, long_decimal_discount, short_decimal_discount, " +
"is_returned, float_quantity, timestamp_shipdate, tinyint_shipmode " +
"FROM lineitem_aria WHERE linenumber = 4";
assertQuery(ariaSession(), query, LINEITEM_ARIA_H2_CTE + query);
The above test reproduces the issue as expected:
Caused by: com.facebook.presto.orc.OrcCorruptionException: Malformed ORC file. Reading BigInteger past EOF [file:/Users/mbasmanova/aria/test_data3/hive_data/tpch/lineitem_aria/20190314_055037_00008_zuqv6_afaded0a-76a8-435e-a41c-dd582fa33ca3]
at com.facebook.presto.orc.stream.DecimalInputStream.skip(DecimalInputStream.java:165)
at com.facebook.presto.orc.reader.DecimalStreamReader.scan(DecimalStreamReader.java:309)
at com.facebook.presto.orc.ColumnGroupReader.advance(ColumnGroupReader.java:413)
at com.facebook.presto.orc.OrcRecordReader.getNextPage(OrcRecordReader.java:828)
at com.facebook.presto.hive.orc.OrcPageSource.getNextPage(OrcPageSource.java:150)
... 13 more
For queries that need to access complex types (arrays, maps, structs), we could compare Hive with Tpch, e.g.
private static final String LINEITEM_ARIA_SQL = "SELECT\n" +
" orderkey,\n" +
" partkey,\n" +
" suppkey,\n" +
" linenumber,\n" +
" quantity,\n" +
" extendedprice,\n" +
" shipmode,\n" +
" shipdate,\n" +
" comment,\n" +
" returnflag = 'R' AS is_returned,\n" +
" CAST(quantity + 1 AS REAL) AS float_quantity,\n" +
" CAST(discount AS decimal(5, 2)) AS short_decimal_discount,\n" +
" CAST(discount AS decimal(38, 2)) AS long_decimal_discount,\n" +
" CAST(array_position(array['SHIP','REG AIR','AIR','FOB','MAIL','RAIL','TRUCK'], shipmode) AS tinyint) AS tinyint_shipmode,\n" +
" date_add('second', suppkey, cast(shipdate as timestamp)) AS timestamp_shipdate,\n" +
" MAP(\n" +
" ARRAY[1, 2, 3],\n" +
" ARRAY[orderkey, partkey, suppkey]\n" +
" ) AS order_part_supp_map,\n" +
" ARRAY[ARRAY[orderkey, partkey, suppkey]] AS order_part_supp_array\n" +
"FROM tpch.tiny.lineitem";
private static final String LINEITEM_ARIA_CTE = format("WITH lineitem_aria AS (%s)\n", LINEITEM_ARIA_SQL);
Session tpchSession = Session.builder(getQueryRunner().getDefaultSession())
.setCatalog("tpch")
.setSchema("tiny")
.build();
assertQuery(ariaSession(), query, tpchSession, LINEITEM_ARIA_CTE + query, "SELECT 1");
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub <#12721 (review)> , or mute the thread <https://github.com/notifications/unsubscribe-auth/AKPPPT6JDW2POSCH3BMP223PSMOFXANCNFSM4HIAA2VQ> .
|
|
@oerling Orri,
The amount of hard-coded SQL is the same in either approach. We need to describe the dataset once anyway. Once it is described the test is just a single assert + query. |
Evaluates filters and filter functions dependent on constant columns at setup time. Supports filter functions on combinations of constant and non-constant columns. Adds constant columns in OrcRecordReader.resultPage(). Fills missing fields of structs with nulls. Prunes the initial qualifying set in OrcRecordReader with any non-deterministic constant filters, evaluating these for every row. Adds a assertQuery signature for comparing Presto execution with different session settings.
|
I mean that the SQL should not reside in a java file to begin with. Like this you get zero cost test development and no mvn and no rebuild.
Sent: Friday, April 26, 2019 12:16 PM
To: prestodb/presto <presto@noreply.github.com>
Cc: oerling <erling@xs4all.nl>; Mention <mention@noreply.github.com>
Subject: Re: [prestodb/presto] Fix skipping in diverse scalar stream readers (#12721)
@oerling <https://github.com/oerling> Orri,
I just want a formulation that does not involve 50 line SQL statements as string literals.
The amount of hard-coded SQL is the same in either approach. We need to describe the dataset once anyway. Once it is described the test is just a single assert + query.
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub <#12721 (comment)> , or mute the thread <https://github.com/notifications/unsubscribe-auth/AKPPPT2SI4VKD45UYGGTNHTPSNIIJANCNFSM4HIAA2VQ> .
|
Resets the toSkip variable after skipping to the next row in the input qualifying set. The error does not manifest if all rows are selected. Sets the expected output size according to the input qualifying set. Adds a test. Note that the test is not supported by H2 and requires comparing Presto with and without Aria scan.
tdcmeehan
reviewed
May 3, 2019
| @@ -292,6 +294,10 @@ public void testFilters() | |||
|
|
|||
| assertQuery(ariaSession(), "SELECT shipdate FROM lineitem_aria WHERE long_decimal_discount < decimal '0.3'", | |||
| "SELECT shipdate FROM lineitem where discount < 0.3"); | |||
| // Select different scalars after a filter. | |||
| assertQuery(ariaSession(), "select partkey, shipmode, comment,\n" + | |||
There was a problem hiding this comment.
Suggested change
| assertQuery(ariaSession(), "select partkey, shipmode, comment,\n" + | |
| assertQuery(ariaSession(), "SELECT partkey, shipmode, comment, " + |
| boolean anyConstantFilterFunctions = false; | ||
| ImmutableList.Builder<FilterFunction> nonDeterministicBuilder = ImmutableList.builder(); | ||
| for (FilterFunction filter : filterFunctions) { | ||
| if (qualifyingSet == null) { |
There was a problem hiding this comment.
Could we initialize this in the constructor?
| public Page getNextPage() | ||
| throws IOException | ||
| { | ||
| if (constantFilterIsFalse) { |
There was a problem hiding this comment.
It would be great if we could make explicit in the code the invariant that pushdownFilterAndProjection is always called before getNextPage.
mbasmanova
force-pushed
the
aria-scan-research
branch
from
ce07e9a
to
60d5dd5
Compare
mbasmanova
force-pushed
the
aria-scan-research
branch
from
053ffb3
to
7c6a815
Compare
mbasmanova
force-pushed
the
aria-scan-research
branch
from
1b8fa52
to
45c09fa
Compare
|
Closing stale PR on behalf of @oerling. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Resets the toSkip variable after skipping to the next row in the input
qualifying set. The error does not manifest if all rows are selected.
Adds a test. Note that the test is not supported by H2 and requires comparing Presto with and without Aria scan.