Looks to me like you found a bug. I think the scanners should be checking both cancellation conditions, i.e. RuntimeState::is_cancelled_ for MT and non-MT scanners and hdfs_scan_node::done_ for non-MT scanners.
On Tue, Jan 16, 2018 at 2:48 PM, Quanlong Huang <huang_quanl...@126.com> wrote: > Hi Tim, > > Thanks for your reply! I have a further question. When given MT_DOP=0, why > don't we use RuntimeState::is_cancelled() to detect cancellation in hdfs > scanners? For example, use it in the loop of ProcessSplit. > There might be a scenario that the FragementInstance was canceled, but the > scanner still don't know about it and then go ahead and pass up all the row > batches. If the FragementInstance just consists of an HdfsScanNode, the > DataStreamSender will try to send these row batches to the upstream > FragmentInstance which has been cancelled. Apparently it'll fail but it > will retry for 2 minutes (in default). The memory resources kept by the > DataStreamSender cannot be released in this 2 minutes window, which might > cause other queries in parallel raising MemLimitExceeded error. > > For example, the plan of query "select 1 from alltypessmall a join > alltypessmall b on a.id != b.id" is > +------------------------------------------------------------------------------------+ > | Max Per-Host Resource Reservation: Memory=0B > | > | Per-Host Resource Estimates: Memory=2.06GB > | > | WARNING: The following tables are missing relevant table and/or column > statistics. | > | functional_orc.alltypessmall > | > | > | > | F02:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1 > | > | Per-Host Resources: mem-estimate=0B mem-reservation=0B > | > | PLAN-ROOT SINK > | > | | mem-estimate=0B mem-reservation=0B > | > | | > | > | 04:EXCHANGE [UNPARTITIONED] > | > | mem-estimate=0B mem-reservation=0B > | > | tuple-ids=0,1 row-size=8B cardinality=unavailable > | > | > | > | F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 > | > | Per-Host Resources: mem-estimate=2.03GB mem-reservation=0B > | > | DATASTREAM SINK [FRAGMENT=F02, EXCHANGE=04, UNPARTITIONED] > | > | | mem-estimate=0B mem-reservation=0B > | > | 02:NESTED LOOP JOIN [INNER JOIN, BROADCAST] > | > | | predicates: a.id != b.id > | > | | mem-estimate=2.00GB mem-reservation=0B > | > | | tuple-ids=0,1 row-size=8B cardinality=unavailable > | > | | > | > | |--03:EXCHANGE [BROADCAST] > | > | | mem-estimate=0B mem-reservation=0B > | > | | tuple-ids=1 row-size=4B cardinality=unavailable > | > | | > | > | 00:SCAN HDFS [functional_orc.alltypessmall a, RANDOM] > | > | partitions=4/4 files=4 size=4.82KB > | > | stored statistics: > | > | table: rows=unavailable size=unavailable > | > | partitions: 0/4 rows=unavailable > | > | columns: unavailable > | > | extrapolated-rows=disabled > | > | mem-estimate=32.00MB mem-reservation=0B > | > | tuple-ids=0 row-size=4B cardinality=unavailable > | > | > | > | F01:PLAN FRAGMENT [RANDOM] hosts=3 instances=3 > | > | Per-Host Resources: mem-estimate=32.00MB mem-reservation=0B > | > | DATASTREAM SINK [FRAGMENT=F00, EXCHANGE=03, BROADCAST] > | > | | mem-estimate=0B mem-reservation=0B > | > | 01:SCAN HDFS [functional_orc.alltypessmall b, RANDOM] > | > | partitions=4/4 files=4 size=4.82KB > | > | stored statistics: > | > | table: rows=unavailable size=unavailable > | > | partitions: 0/4 rows=unavailable > | > | columns: unavailable > | > | extrapolated-rows=disabled > | > | mem-estimate=32.00MB mem-reservation=0B > | > | tuple-ids=1 row-size=4B cardinality=unavailable > | > +------------------------------------------------------------------------------------+ > > When errors happen in F00, cancellation rpc will be sent to F01. However, the > hdfs scanner in F01 does not notice it in time and pass up all the row > batches. Then the DataStreamSender will try to send these row batches to F01. > It will retry for 2 minutes. In this time window it might hold significant > memory resources, which causes other queries cannot allocate memory and fail. > This can be avoid if the hdfs scanner use RuntimeState::is_cancelled() to > detect the cancellation in time. > > Am I right? > > Thanks, > Quanlong > > At 2018-01-17 01:05:57, "Tim Armstrong" <tarmstr...@cloudera.com> wrote: > >ScannerContext::cancelled() == true means that the scan has completed, > >either because it has returned enough rows, because the query is cancelled, > >or because it hit an error. > > > >RuntimeState::cancelled() == true means that the query is cancelled. > > > >So there are cases where ScannerContext::cancelled() == true and > >RuntimeState::cancelled() is false. E.g. where there's a limit on the scan. > > > >I think the name of ScannerContext::cancelled() is misleading, it should > >probably be called "done()" to match HdfsScanNode::done(). More generally, > >the cancellation logic could probably be cleaned up and simplified further. > > > >On Mon, Jan 15, 2018 at 6:20 PM, Quanlong Huang <huang_quanl...@126.com> > >wrote: > > > >> Hi all, > >> > >> > >> I'm confused about the cancellation logic in hdfs scanners. There're two > >> functions to detect cancellation: ScannerContext::cancelled() and > >> RuntimeState::is_cancelled(). > >> When MT_DOP is not set (i.e. MT_DOP=0), ScannerContext::cancelled() will > >> return HdfsScanNode::done(). However, the field done_ in HdfsScanNode seems > >> to be set according to status return from scanners. > >> I've witnessed some points when RuntimeState::is_cancelled() is true but > >> ScannerContext::cancelled() is false. > >> > >> > >> My question is why scanners don't use RuntimeState::is_cancelled() to > >> detect cancellation, which is more timely than using > >> ScannerContext::cancelled(). There must be some detailed reasons that I've > >> missed. Would you be so kind to answer my question? > >> > >> > >> Thanks, > >> Quanlong > > > > >
