Propagate stats and cost in distributed explain plans

[rschlussel]

fixes #10963

[kokosing]

planFragments?

[rschlussel]

Changing to planFragmentsMap. I don't like when maps are named without the word map (or something that makes it clear it's a mapping). I always expect them to be a list or a set (or any other collection of just one thing).

[martint]

Heh... I actually dislike when variable names encode their type (unless there's a very good reason for that, such as a conflict with some other variable -- e.g., a when a builder is used, or when having to convert types and you need to hold on to both versions). In this particular case, fragmentPlanNodeMap or planFragmentsMap doesn't even convey what it represents. It's even ambiguous, since the main entity it holds is plan nodes (indexed by which fragment they belong to). I would suggest sourcesByFragment, nodesByFragment, sourceNodesByFragment, fragmentToSource or something like that.

[kokosing]

s/createFragmentedPlanSourceProvider/create

Also I would put it at the top, above the constructor

[kokosing]

double licence

[kokosing]

rnn

[kokosing]

should not this method be added as a regular rule to com.facebook.presto.cost.ComposableStatsCalculator?

[kokosing]

Why is this public?

[rschlussel]

I did it that way at first but it doesn't work (at least not neatly) because it needs the state from the specific query to be able to get the sources.

[kokosing]

move it to the top, just below the constructor

[kokosing]

add @FunctionalInterface

[kokosing]

RemoteSourceStatsCalculator?

[rschlussel]

it's a statscalculator for the whole fragmented plan. it just handles a remote source node.

[kokosing]

Inject PlanSourceProvider

[rschlussel]

It needs to be created new for each query.

[kokosing]

No tests for this class?

[rschlussel]

I'll work on that tomorrow

[arhimondr]

indent used to be 2 here. Now, if i'm following correctly, it is 0

[rschlussel]

yeah but it was equally readable as zero, so it didn't seem worth having its own specialized constructor.

[arhimondr]

Move this constructor after the very next one.

[arhimondr]

ImmutableMap.copyOf(requireNonNull(fragmentPlanNodeMap, "fragmentPlanNodeMap is null"))

[arhimondr]

Move it before the constructor

[arhimondr]

null check

[arhimondr]

verify(estimate!=null, "estimate is absent")

[arhimondr]

Keep it after calculateRemoteSourceStats after you move one

[arhimondr]

I like functional programming. But this seems like an abuse of one. I think something like this is way more readable

if(estimate != null){
  estmate = addStatsAndMaxDistinctValues(estmate, currentSourceEstimate)
} else {
  estimate = currentSourceEstimate;
}

[arhimondr]

PlanNodeStatsEstimate currentSourceEstimate = mapToOutputSymbols(statsProvider.getStats(source), source.getOutputSymbols(), node.getOutputSymbols())

[arhimondr]

allFragments

[rschlussel]

Addressed comments except writing tests and updated how PlanFragmentStatsCalculator gets the input symbols to fix the failing explain analyze tests.

[kokosing]

Maybe FragmentedPlanSourceProvider::getSources for RemoteSourceNode should return nodes with output symbols already mapped? That way you wouldn't need to map symbols here.

[rschlussel]

i think that would mess up getting the stats for those sources.

[findepi]

Add a comment here explaining why remote sources are ignored in cost estimation.

of course, throwing would be safer here, in case cost calculator is used on actual plan somewhere else than in EXPLAIN.

[rschlussel]

I was just making this a pass through so we could get the costs for other things, but now I'm thinking that we'll lose the exchange cost. Maybe I should make a FragmentCostCalculator extension to get the exchange properties from the fragment and then call calculateExchangeCost

[findepi]

What about different approach -- that we de-fragment the plan and pass that to the cost calculator? Then the stats & costing code would not need to be modified.

[findepi]

I was just making this a pass through so we could get the costs for other things, but now I'm thinking that we'll lose the exchange cost.

I didn't notice. The problem is that it's possible for "cost on fragmented plan" to be very off, and it may go unnoticed for long time: are we able to test-cover everything? will we notice in production when costs are off? (ultimately we will, but not too soon, as the costing here won't have any impact on execution)

[rschlussel]

I had looked at defragmenting and computing the cost, but it gets very messy in the plan printer because each node you visit needs to have the defragmented plan and match up correctly (or re-defragment), etc.

I'm going to go with the FragmentedPlanCostCalculator approach for handling RemoteSourceNodes and I'll add a test that creates a plan and computes the cost and then fragments the plan and checks that the cumulative cost is the same.

[findepi]

Can sourceProvider be modelled with a lookup?

[rschlussel]

not without abusing the interface :). Lookups are for resolving GroupReferences to the nodes that are a part of that group. A SourceProvider gives you the list of source nodes for a given node.

[findepi]

Did you check this PR for potential conflicts with #11267?

[rschlussel]

doesn't overlap i don't think.

[rschlussel]

Updated:

1. Added a commit to include stats in plans printed from QueryMonitor.
2. @haozhun beat me at getting rid of all the overloaded methods in PlanPrinter, so from that commit I now just have a small simplification in formatFragment.
3. Added support for computing "exchange costs" for remote source nodes (which was a bit of an invasive change) and added tests for the fragmented plan cost and stats calculators

[rschlussel]

Also @arhimondr. i didn't get rid of the stats calculator interface because it's useful for testing.

[arhimondr]

LGTM % COMMENTS

[arhimondr]

final

[arhimondr]

final

[arhimondr]

nit: i would make the delegate field to be the first field and the first parameter in the constructor. You can do this easily by applying automatic refactoring.

[arhimondr]

This seems to be fragile. How about storing the exchange type in the RemoteSourceNode? It should be an easy change.

[arhimondr]

You can simply call it fragments. There is no name clash.

[arhimondr]

message

[arhimondr]

This config is being created twice in this class. Create in in the LocalQueryRunner constructor and store it as a field, to make sure it is always the same.

[arhimondr]

• Nullify all the fields after test class is finished
• Use createTestTransactionManager(catalogManager)
• Add constructor method that accepts transactionManager in MetadataManager, and use createTestMetadataManager(new FeaturesConfig(), transactionManager)
• Optionally you can create similar createTest* methods for nodePartitioningManager, that creates test nodePartitioningManager that isn't required to be stopped, and can be simply nullified (uses direct executors instead of thread pools, etc.)

Here is the example:

private CostCalculator costCalculatorUsingExchanges;
    private CostCalculator costCalculatorWithEstimatedExchanges;
    private PlanFragmenter planFragmenter;
    private Session session;
    private MetadataManager metadata;
    private TransactionManager transactionManager;
    private FinalizerService finalizerService;
    private NodeScheduler nodeScheduler;
    private NodePartitioningManager nodePartitioningManager;

    @BeforeClass
    public void setUp()
    {
        costCalculatorUsingExchanges = new CostCalculatorUsingExchanges(() -> NUMBER_OF_NODES);
        costCalculatorWithEstimatedExchanges = new CostCalculatorWithEstimatedExchanges(costCalculatorUsingExchanges, () -> NUMBER_OF_NODES);
        planFragmenter = new PlanFragmenter(new QueryManagerConfig());

        session = testSessionBuilder().setCatalog("tpch").build();

        CatalogManager catalogManager = new CatalogManager();
        catalogManager.registerCatalog(createBogusTestingCatalog("tpch"));
        transactionManager = createTestTransactionManager(catalogManager);
        metadata = createTestMetadataManager(new FeaturesConfig(), transactionManager);

        finalizerService = new FinalizerService();
        finalizerService.start();
        nodeScheduler = new NodeScheduler(
                new LegacyNetworkTopology(),
                new InMemoryNodeManager(),
                new NodeSchedulerConfig().setIncludeCoordinator(true),
                new NodeTaskMap(finalizerService));
        nodePartitioningManager = new NodePartitioningManager(nodeScheduler);
    }

    @AfterClass(alwaysRun = true)
    public void tearDown()
    {
        costCalculatorUsingExchanges = null;
        costCalculatorWithEstimatedExchanges = null;
        planFragmenter = null;
        session = null;
        transactionManager = null;
        metadata = null;
        finalizerService.destroy();
        finalizerService = null;
        nodeScheduler.stop();
        nodeScheduler = null;
        nodePartitioningManager = null;
    }

[arhimondr]

Similar comment about bootstraping the test class

private Metadata metadata;
    private Session session;

    @BeforeMethod
    public void setUp()
    {
        metadata = createTestMetadataManager();
        session = testSessionBuilder().build();
    }

    @AfterMethod(alwaysRun = true)
    public void tearDown()
    {
        metadata = null;
        session = null;
    }

[arhimondr]

Remove this method

[rschlussel]

Thanks for the review! Updated and I'll merge on Monday.

[wenleix]

@rschlussel : My understanding is we need allFragments as we want to create sourceProvider and statsCalculator. However, later you make stats and cost precomputed and store in PlanFragment: #11511

I am wondering do we still need to instantiate the TypeProvider with all fragments?