You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "gargvishesh (via GitHub)" <gi...@apache.org> on 2023/10/20 18:01:14 UTC
[PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
gargvishesh opened a new pull request, #15222:
URL: https://github.com/apache/druid/pull/15222
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Fixes #14514
### Description
Segregate `advance` and `advanceUninterruptibly` flow in `postJoinCursor` by calling `baseCursor.advance` and `baseCursor.advanceUninterruptibly` in them, respectively.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1374082055
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
nit : catch clause can be removed and exception declared on method.
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
Review Comment:
shouldn't this return false? May not matter but given its infinite cursor.
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
Review Comment:
A 10 second test is expensive. Rather, we should use a count latch or something to wait for X records to be processed. The latch wait can have 5 second timeout. Similarly, the 5 second sleep after interrupting should be removed and we can wait in a loop or use count latch again in ExceptionHandler.
I also like https://github.com/awaitility/awaitility/wiki/Usage#simple in case you want to bring that in here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1372965865
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
Review Comment:
## Field masks field in super class
This field shadows another field called [factSegment](1) in a superclass.
[Show more details](https://github.com/apache/druid/security/code-scanning/5935)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "pranavbhole (via GitHub)" <gi...@apache.org>.
pranavbhole commented on PR #15222:
URL: https://github.com/apache/druid/pull/15222#issuecomment-1773185553
can we add test for this scenario if possible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on PR #15222:
URL: https://github.com/apache/druid/pull/15222#issuecomment-1773274538
```
Error: /home/runner/work/druid/druid/processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:22:8: Unused import - org.apache.druid.query.BaseQuery. [UnusedImports]
```
My intellij doesn't clean up unused imports automatically. I run `mvn checkstyle:checkstyle` to confirm that checkstyle is passing. Sometimes deeply nested stuff won't get flagged up in Intellij but will show up as a check style failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "gargvishesh (via GitHub)" <gi...@apache.org>.
gargvishesh commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1375254805
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
Review Comment:
Went with a latch, but thanks for the pointer to this library.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1375666863
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment infiniteFactSegment;
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ CountDownLatch countDownLatch;
+
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index, CountDownLatch countDownLatch)
+ {
+ super(index);
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics)
+ .map(cursor -> new CursorNoAdvance(cursor, countDownLatch));
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+ CountDownLatch countDownLatch;
+
+ public CursorNoAdvance(Cursor cursor, CountDownLatch countDownLatch)
+ {
+ this.cursor = cursor;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ countDownLatch.countDown();
+
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId, CountDownLatch countDownLatch)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index, countDownLatch);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption() throws IOException, InterruptedException
+ {
+
+ final int rowsBeforeInterrupt = 1000;
+
+ CountDownLatch countDownLatch = new CountDownLatch(rowsBeforeInterrupt);
+
+ infiniteFactSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts"),
+ countDownLatch
+ );
+
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+
+ countDownLatch.await(1, TimeUnit.SECONDS);
+ joinCursorThread.interrupt();
+
+ // Wait for a max of 1 sec for the exception to be set.
+ for (int i = 0; i < 1000; i++) {
+ if (exceptionHandler.getException() == null) {
+ sleep(1);
+ } else {
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
Review Comment:
you should just return the call here and also add `assertFalse` at the end of the method. So we are checking that we did get QueryInterruptedException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "gargvishesh (via GitHub)" <gi...@apache.org>.
gargvishesh commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1372916576
##########
processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:
##########
@@ -70,6 +70,15 @@ public static PostJoinCursor wrap(
}
private void advanceToMatch()
+ {
+ if (valueMatcher != null) {
+ while (!isDone() && !valueMatcher.matches(false)) {
+ baseCursor.advance();
+ }
+ }
+ }
+
+ private void advanceToMatchUninterruptibly()
Review Comment:
Done
##########
processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:
##########
@@ -99,15 +108,15 @@ public Filter getPostJoinFilter()
@Override
public void advance()
{
- advanceUninterruptibly();
- BaseQuery.checkInterrupted();
Review Comment:
Added a comment.
##########
processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:
##########
@@ -99,15 +108,15 @@ public Filter getPostJoinFilter()
@Override
public void advance()
{
- advanceUninterruptibly();
- BaseQuery.checkInterrupted();
+ baseCursor.advance();
Review Comment:
Added comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1374834656
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
+ }
+
+ public void makeCursorAndAdvance()
+ {
+
+ List<JoinableClause> joinableClauses = ImmutableList.of(
+ factToCountryOnIsoCode(JoinType.LEFT)
+ );
+
+ JoinFilterPreAnalysis joinFilterPreAnalysis = makeDefaultConfigPreAnalysis(
+ null,
+ joinableClauses,
+ VirtualColumns.EMPTY
+ );
+
+ HashJoinSegmentStorageAdapter hashJoinSegmentStorageAdapter = new HashJoinSegmentStorageAdapter(
+ factSegment.asStorageAdapter(),
+ joinableClauses,
+ joinFilterPreAnalysis
+ );
+
+ Cursor cursor = Iterables.getOnlyElement(hashJoinSegmentStorageAdapter.makeCursors(
+ null,
+ Intervals.ETERNITY,
+ VirtualColumns.EMPTY,
+ Granularities.ALL,
+ false,
+ null
+ ).toList());
+
+ ((PostJoinCursor)cursor).setValueMatcher(new ValueMatcher()
+ {
+ @Override
+ public boolean matches(boolean includeUnknown)
+ {
+ return false;
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+
+ }
+ });
+
+ cursor.advance();
+ }
+}
Review Comment:
Yeah it is needed, the checkstyle is complaining. It is good to run a `mvn checkstyle:checkstyle` beforehand as Laksh have suggested earlier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor to allow for interrupts in advance (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 merged PR #15222:
URL: https://github.com/apache/druid/pull/15222
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1374075989
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
Review Comment:
Please add the license header here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1367368051
##########
processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:
##########
@@ -99,15 +108,15 @@ public Filter getPostJoinFilter()
@Override
public void advance()
{
- advanceUninterruptibly();
- BaseQuery.checkInterrupted();
+ baseCursor.advance();
Review Comment:
Will be good to put some comments as in why this change is being done in the code for a future developer to follow.
##########
processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:
##########
@@ -99,15 +108,15 @@ public Filter getPostJoinFilter()
@Override
public void advance()
{
- advanceUninterruptibly();
- BaseQuery.checkInterrupted();
Review Comment:
Can we add a note on why the checkInterrupted() is not needed anymore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "gargvishesh (via GitHub)" <gi...@apache.org>.
gargvishesh commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1375673257
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment infiniteFactSegment;
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ CountDownLatch countDownLatch;
+
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index, CountDownLatch countDownLatch)
+ {
+ super(index);
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics)
+ .map(cursor -> new CursorNoAdvance(cursor, countDownLatch));
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+ CountDownLatch countDownLatch;
+
+ public CursorNoAdvance(Cursor cursor, CountDownLatch countDownLatch)
+ {
+ this.cursor = cursor;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ countDownLatch.countDown();
+
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId, CountDownLatch countDownLatch)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index, countDownLatch);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption() throws IOException, InterruptedException
+ {
+
+ final int rowsBeforeInterrupt = 1000;
+
+ CountDownLatch countDownLatch = new CountDownLatch(rowsBeforeInterrupt);
+
+ infiniteFactSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts"),
+ countDownLatch
+ );
+
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+
+ countDownLatch.await(1, TimeUnit.SECONDS);
+ joinCursorThread.interrupt();
+
+ // Wait for a max of 1 sec for the exception to be set.
+ for (int i = 0; i < 1000; i++) {
+ if (exceptionHandler.getException() == null) {
+ sleep(1);
+ } else {
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
Review Comment:
Makes sense. Done - thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1368822377
##########
processing/src/main/java/org/apache/druid/segment/join/PostJoinCursor.java:
##########
@@ -70,6 +70,15 @@ public static PostJoinCursor wrap(
}
private void advanceToMatch()
+ {
+ if (valueMatcher != null) {
+ while (!isDone() && !valueMatcher.matches(false)) {
+ baseCursor.advance();
+ }
+ }
+ }
+
+ private void advanceToMatchUninterruptibly()
Review Comment:
Please leave a comment here that this can be a long-running CPU call. Which is why advanceUninterruptibly is not directly used in advance() call unlike other cursors. Please link the github issue in the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1374077155
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
+ }
+
+ public void makeCursorAndAdvance()
+ {
+
+ List<JoinableClause> joinableClauses = ImmutableList.of(
+ factToCountryOnIsoCode(JoinType.LEFT)
+ );
+
+ JoinFilterPreAnalysis joinFilterPreAnalysis = makeDefaultConfigPreAnalysis(
+ null,
+ joinableClauses,
+ VirtualColumns.EMPTY
+ );
+
+ HashJoinSegmentStorageAdapter hashJoinSegmentStorageAdapter = new HashJoinSegmentStorageAdapter(
+ factSegment.asStorageAdapter(),
+ joinableClauses,
+ joinFilterPreAnalysis
+ );
+
+ Cursor cursor = Iterables.getOnlyElement(hashJoinSegmentStorageAdapter.makeCursors(
+ null,
+ Intervals.ETERNITY,
+ VirtualColumns.EMPTY,
+ Granularities.ALL,
+ false,
+ null
+ ).toList());
+
+ ((PostJoinCursor)cursor).setValueMatcher(new ValueMatcher()
+ {
+ @Override
+ public boolean matches(boolean includeUnknown)
+ {
+ return false;
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+
+ }
+ });
+
+ cursor.advance();
+ }
+}
Review Comment:
I think we need to leave a blank line at the end of each file. Not sure though
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on PR #15222:
URL: https://github.com/apache/druid/pull/15222#issuecomment-1773209952
I dunno why the checkstyle is failing though. If using IntelliJ it'll be good to add this https://raw.githubusercontent.com/apache/incubator-druid/master/dev/druid_intellij_formatting.xml to IntelliJ
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "gargvishesh (via GitHub)" <gi...@apache.org>.
gargvishesh commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1375254640
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
+ }
+
+ public void makeCursorAndAdvance()
+ {
+
+ List<JoinableClause> joinableClauses = ImmutableList.of(
+ factToCountryOnIsoCode(JoinType.LEFT)
+ );
+
+ JoinFilterPreAnalysis joinFilterPreAnalysis = makeDefaultConfigPreAnalysis(
+ null,
+ joinableClauses,
+ VirtualColumns.EMPTY
+ );
+
+ HashJoinSegmentStorageAdapter hashJoinSegmentStorageAdapter = new HashJoinSegmentStorageAdapter(
+ factSegment.asStorageAdapter(),
+ joinableClauses,
+ joinFilterPreAnalysis
+ );
+
+ Cursor cursor = Iterables.getOnlyElement(hashJoinSegmentStorageAdapter.makeCursors(
+ null,
+ Intervals.ETERNITY,
+ VirtualColumns.EMPTY,
+ Granularities.ALL,
+ false,
+ null
+ ).toList());
+
+ ((PostJoinCursor)cursor).setValueMatcher(new ValueMatcher()
+ {
+ @Override
+ public boolean matches(boolean includeUnknown)
+ {
+ return false;
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+
+ }
+ });
+
+ cursor.advance();
+ }
+}
Review Comment:
Shall follow. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "gargvishesh (via GitHub)" <gi...@apache.org>.
gargvishesh commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1375254725
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "gargvishesh (via GitHub)" <gi...@apache.org>.
gargvishesh commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1375254684
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
Review Comment:
Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1374834656
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
+ }
+
+ public void makeCursorAndAdvance()
+ {
+
+ List<JoinableClause> joinableClauses = ImmutableList.of(
+ factToCountryOnIsoCode(JoinType.LEFT)
+ );
+
+ JoinFilterPreAnalysis joinFilterPreAnalysis = makeDefaultConfigPreAnalysis(
+ null,
+ joinableClauses,
+ VirtualColumns.EMPTY
+ );
+
+ HashJoinSegmentStorageAdapter hashJoinSegmentStorageAdapter = new HashJoinSegmentStorageAdapter(
+ factSegment.asStorageAdapter(),
+ joinableClauses,
+ joinFilterPreAnalysis
+ );
+
+ Cursor cursor = Iterables.getOnlyElement(hashJoinSegmentStorageAdapter.makeCursors(
+ null,
+ Intervals.ETERNITY,
+ VirtualColumns.EMPTY,
+ Granularities.ALL,
+ false,
+ null
+ ).toList());
+
+ ((PostJoinCursor)cursor).setValueMatcher(new ValueMatcher()
+ {
+ @Override
+ public boolean matches(boolean includeUnknown)
+ {
+ return false;
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+
+ }
+ });
+
+ cursor.advance();
+ }
+}
Review Comment:
Yeah it is needed, the checkstyle is complaining. It is good to run a `mvn checkstyle:checkstyle` beforehand as Laksh have suggested earlier. Another thing I do is in IntelliJ when adding a new section of code have the [druid coding style](https://raw.githubusercontent.com/apache/incubator-druid/master/dev/druid_intellij_formatting.xml) and run a reformat code that fixes most of the codestyling for me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Segregate advance and advanceUninterruptibly flow in postJoinCursor (druid)
Posted by "soumyava (via GitHub)" <gi...@apache.org>.
soumyava commented on code in PR #15222:
URL: https://github.com/apache/druid/pull/15222#discussion_r1374077155
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -0,0 +1,230 @@
+package org.apache.druid.segment.join;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.QueryInterruptedException;
+import org.apache.druid.query.QueryMetrics;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.QueryableIndexStorageAdapter;
+import org.apache.druid.segment.StorageAdapter;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.join.filter.JoinFilterPreAnalysis;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.Thread.sleep;
+import static org.junit.Assert.assertTrue;
+
+public class PostJoinCursorTest extends BaseHashJoinSegmentStorageAdapterTest
+{
+
+ public QueryableIndexSegment factSegment;
+
+ @Override
+ public void setUp() throws IOException
+ {
+
+ factSegment = new TestInfiniteQueryableIndexSegment(
+ JoinTestHelper.createFactIndexBuilder(temporaryFolder.newFolder()).buildMMappedIndex(),
+ SegmentId.dummy("facts")
+ );
+ countryIsoCodeToNameLookup = JoinTestHelper.createCountryIsoCodeToNameLookup();
+ countryNumberToNameLookup = JoinTestHelper.createCountryNumberToNameLookup();
+ countriesTable = JoinTestHelper.createCountriesIndexedTable();
+ regionsTable = JoinTestHelper.createRegionsIndexedTable();
+ }
+
+ /**
+ * Simulates infinite segment by using a base cursor with advance() and advanceInterruptibly()
+ * reduced to a no-op.
+ */
+ private static class TestInfiniteQueryableIndexSegment extends QueryableIndexSegment
+ {
+
+ private final StorageAdapter testStorageAdaptor;
+
+ public TestInfiniteQueryableIndexSegment(QueryableIndex index, SegmentId segmentId)
+ {
+ super(index, segmentId);
+ testStorageAdaptor = new InfiniteQueryableIndexStorageAdapter(index);
+ }
+
+ @Override
+ public StorageAdapter asStorageAdapter()
+ {
+ return testStorageAdaptor;
+ }
+ }
+
+ private static class InfiniteQueryableIndexStorageAdapter extends QueryableIndexStorageAdapter
+ {
+ public InfiniteQueryableIndexStorageAdapter(QueryableIndex index)
+ {
+ super(index);
+ }
+
+ @Override
+ public Sequence<Cursor> makeCursors(
+ @Nullable Filter filter,
+ Interval interval,
+ VirtualColumns virtualColumns,
+ Granularity gran,
+ boolean descending,
+ @Nullable QueryMetrics<?> queryMetrics
+ )
+ {
+ return super.makeCursors(filter, interval, virtualColumns, gran, descending, queryMetrics).map(
+ CursorNoAdvance::new);
+ }
+
+ private static class CursorNoAdvance implements Cursor
+ {
+ Cursor cursor;
+
+ public CursorNoAdvance(Cursor cursor)
+ {
+ this.cursor = cursor;
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return cursor.getColumnSelectorFactory();
+ }
+
+ @Override
+ public DateTime getTime()
+ {
+ return cursor.getTime();
+ }
+
+ @Override
+ public void advance()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ // Do nothing to simulate infinite rows
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return cursor.isDone();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return cursor.isDoneOrInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+
+ }
+ }
+ }
+
+ private static class ExceptionHandler implements Thread.UncaughtExceptionHandler
+ {
+
+ Throwable exception;
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e)
+ {
+ exception = e;
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+ }
+
+ @Test
+ public void testAdvanceWithInterruption()
+ {
+ Thread joinCursorThread = new Thread(() -> makeCursorAndAdvance());
+ ExceptionHandler exceptionHandler = new ExceptionHandler();
+ joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
+ joinCursorThread.start();
+ try {
+ sleep(5000);
+ joinCursorThread.interrupt();
+ sleep(5000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ assertTrue(exceptionHandler.getException() instanceof QueryInterruptedException);
+ }
+
+ public void makeCursorAndAdvance()
+ {
+
+ List<JoinableClause> joinableClauses = ImmutableList.of(
+ factToCountryOnIsoCode(JoinType.LEFT)
+ );
+
+ JoinFilterPreAnalysis joinFilterPreAnalysis = makeDefaultConfigPreAnalysis(
+ null,
+ joinableClauses,
+ VirtualColumns.EMPTY
+ );
+
+ HashJoinSegmentStorageAdapter hashJoinSegmentStorageAdapter = new HashJoinSegmentStorageAdapter(
+ factSegment.asStorageAdapter(),
+ joinableClauses,
+ joinFilterPreAnalysis
+ );
+
+ Cursor cursor = Iterables.getOnlyElement(hashJoinSegmentStorageAdapter.makeCursors(
+ null,
+ Intervals.ETERNITY,
+ VirtualColumns.EMPTY,
+ Granularities.ALL,
+ false,
+ null
+ ).toList());
+
+ ((PostJoinCursor)cursor).setValueMatcher(new ValueMatcher()
+ {
+ @Override
+ public boolean matches(boolean includeUnknown)
+ {
+ return false;
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+
+ }
+ });
+
+ cursor.advance();
+ }
+}
Review Comment:
I think we need to leave a blank line at the end of each file. Not sure though. If checkstyle oks it am good with it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org