#9967

Committed 30 Aug 2023 04:22PM CUT coverage: 47.7% (+0.04%) from 47.658%

Build # #9967

Build Type

push

travis_ci

Committed by

web-flow

Commit Message

Pipe: Fix start-time and end-time parameters not working when extracting history data (#11001) (#11002)

(cherry picked from commit 35736cc67)

Run Details

12 of 12 new or added lines in 6 files covered. (100.0%)

80165 of 168062 relevant lines covered (47.7%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.51

/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/RootProcedureStack.java

/*
 * 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.iotdb.confignode.procedure;

import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.state.ProcedureState;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class RootProcedureStack<Env> {
  private static final Logger LOG = LoggerFactory.getLogger(RootProcedureStack.class);

  private enum State {
    RUNNING, // The Procedure is running or ready to run
    FAILED, // The Procedure failed, waiting for the rollback executing
    ROLLINGBACK, // The Procedure failed and the execution was rolledback
  }

  private Set<Procedure<Env>> subprocs = null;
  private ArrayList<Procedure<Env>> subprocStack = null;
  private State state = State.RUNNING;
  private int running = 0;

  public synchronized boolean isFailed() {
    switch (state) {
      case ROLLINGBACK:
      case FAILED:
        return true;
      default:
        break;
    }
    return false;
  }

  public synchronized boolean isRollingback() {
    return state == State.ROLLINGBACK;
  }

  /** Called by the {@link ProcedureExecutor} to mark rollback execution. */
  protected synchronized boolean setRollback() {
    if (running == 0 && state == State.FAILED) {
      state = State.ROLLINGBACK;
      return true;
    }
    return false;
  }

  /** Called by the {@link ProcedureExecutor} to mark rollback execution. */
  protected synchronized void unsetRollback() {
    assert state == State.ROLLINGBACK;
    state = State.FAILED;
  }

  protected synchronized long[] getSubprocedureIds() {
    if (subprocs == null) {
      return new long[0];
    }
    return subprocs.stream().mapToLong(Procedure::getProcId).toArray();
  }

  protected synchronized List<Procedure<Env>> getSubproceduresStack() {
    return subprocStack;
  }

  protected synchronized ProcedureException getException() {
    if (subprocStack != null) {
      for (Procedure<Env> proc : subprocStack) {
        if (proc.hasException()) {
          return proc.getException();
        }
      }
    }
    return null;
  }

  /** Called by the {@link ProcedureExecutor} to mark the procedure step as running. */
  protected synchronized boolean acquire() {
    if (state != State.RUNNING) {
      return false;
    }

    running++;
    return true;
  }

  /** Called by the {@link ProcedureExecutor} to mark the procedure step as finished. */
  protected synchronized void release() {
    running--;
  }

  protected synchronized void abort() {
    if (state == State.RUNNING) {
      state = State.FAILED;
    }
  }

  /**
   * Called by the {@link ProcedureExecutor} after the procedure step is completed, to add the step
   * to the rollback list (or procedure stack).
   */
  protected synchronized void addRollbackStep(Procedure<Env> proc) {
    if (proc.isFailed()) {
      state = State.FAILED;
    }
    if (subprocStack == null) {
      subprocStack = new ArrayList<>();
    }
    proc.addStackIndex(subprocStack.size());
    LOG.trace("Add procedure {} as the {}th rollback step", proc, subprocStack.size());
    subprocStack.add(proc);
  }

  protected synchronized void addSubProcedure(Procedure<Env> proc) {
    if (!proc.hasParent()) {
      return;
    }
    if (subprocs == null) {
      subprocs = new HashSet<>();
    }
    subprocs.add(proc);
  }

  /**
   * Called on store load by the {@link ProcedureExecutor} to load part of the stack.
   *
   * <p>Each procedure has its own stack-positions. Which means we have to write to the store only
   * the Procedure we executed, and nothing else. On load we recreate the full stack by aggregating
   * each procedure stack-positions.
   */
  protected synchronized void loadStack(Procedure<Env> proc) {
    addSubProcedure(proc);
    int[] stackIndexes = proc.getStackIndexes();
    if (stackIndexes != null) {
      if (subprocStack == null) {
        subprocStack = new ArrayList<>();
      }
      int diff = (1 + stackIndexes[stackIndexes.length - 1]) - subprocStack.size();
      if (diff > 0) {
        subprocStack.ensureCapacity(1 + stackIndexes[stackIndexes.length - 1]);
        while (diff-- > 0) {
          subprocStack.add(null);
        }
      }
      for (int stackIndex : stackIndexes) {
        subprocStack.set(stackIndex, proc);
      }
    }
    if (proc.getState() == ProcedureState.ROLLEDBACK) {
      state = State.ROLLINGBACK;
    } else if (proc.isFailed()) {
      state = State.FAILED;
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one
3	* or more contributor license agreements. See the NOTICE file
4	* distributed with this work for additional information
5	* regarding copyright ownership. The ASF licenses this file
6	* to you under the Apache License, Version 2.0 (the
7	* "License"); you may not use this file except in compliance
8	* with the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing,
13	* software distributed under the License is distributed on an
14	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15	* KIND, either express or implied. See the License for the
16	* specific language governing permissions and limitations
17	* under the License.
18	*/
19
20	package org.apache.iotdb.confignode.procedure;
21
22	import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
23	import org.apache.iotdb.confignode.procedure.state.ProcedureState;
24
25	import org.slf4j.Logger;
26	import org.slf4j.LoggerFactory;
27
28	import java.util.ArrayList;
29	import java.util.HashSet;
30	import java.util.List;
31	import java.util.Set;
32
33	public class RootProcedureStack<Env> {	1✔
34	private static final Logger LOG = LoggerFactory.getLogger(RootProcedureStack.class);	1✔
35
36	private enum State {	1✔
37	RUNNING, // The Procedure is running or ready to run	1✔
38	FAILED, // The Procedure failed, waiting for the rollback executing	1✔
39	ROLLINGBACK, // The Procedure failed and the execution was rolledback	1✔
40	}
41
42	private Set<Procedure<Env>> subprocs = null;	1✔
43	private ArrayList<Procedure<Env>> subprocStack = null;	1✔
44	private State state = State.RUNNING;	1✔
45	private int running = 0;	1✔
46
47	public synchronized boolean isFailed() {
48	switch (state) {	1✔
49	case ROLLINGBACK:
50	case FAILED:
51	return true;	1✔
52	default:
53	break;
54	}
55	return false;	1✔
56	}
57
58	public synchronized boolean isRollingback() {
59	return state == State.ROLLINGBACK;	×
60	}
61
62	/** Called by the {@link ProcedureExecutor} to mark rollback execution. */
63	protected synchronized boolean setRollback() {
64	if (running == 0 && state == State.FAILED) {	1✔
65	state = State.ROLLINGBACK;	1✔
66	return true;	1✔
67	}
68	return false;	×
69	}
70
71	/** Called by the {@link ProcedureExecutor} to mark rollback execution. */
72	protected synchronized void unsetRollback() {
73	assert state == State.ROLLINGBACK;	×
74	state = State.FAILED;	×
75	}	×
76
77	protected synchronized long[] getSubprocedureIds() {
78	if (subprocs == null) {	1✔
79	return new long[0];	1✔
80	}
81	return subprocs.stream().mapToLong(Procedure::getProcId).toArray();	1✔
82	}
83
84	protected synchronized List<Procedure<Env>> getSubproceduresStack() {
85	return subprocStack;	1✔
86	}
87
88	protected synchronized ProcedureException getException() {
89	if (subprocStack != null) {	1✔
90	for (Procedure<Env> proc : subprocStack) {	1✔
91	if (proc.hasException()) {	1✔
92	return proc.getException();	1✔
93	}
94	}	1✔
95	}
96	return null;	×
97	}
98
99	/** Called by the {@link ProcedureExecutor} to mark the procedure step as running. */
100	protected synchronized boolean acquire() {
101	if (state != State.RUNNING) {	1✔
102	return false;	1✔
103	}
104
105	running++;	1✔
106	return true;	1✔
107	}
108
109	/** Called by the {@link ProcedureExecutor} to mark the procedure step as finished. */
110	protected synchronized void release() {
111	running--;	1✔
112	}	1✔
113
114	protected synchronized void abort() {
115	if (state == State.RUNNING) {	×
116	state = State.FAILED;	×
117	}
118	}	×
119
120	/**
121	* Called by the {@link ProcedureExecutor} after the procedure step is completed, to add the step
122	* to the rollback list (or procedure stack).
123	*/
124	protected synchronized void addRollbackStep(Procedure<Env> proc) {
125	if (proc.isFailed()) {	1✔
126	state = State.FAILED;	1✔
127	}
128	if (subprocStack == null) {	1✔
129	subprocStack = new ArrayList<>();	1✔
130	}
131	proc.addStackIndex(subprocStack.size());	1✔
132	LOG.trace("Add procedure {} as the {}th rollback step", proc, subprocStack.size());	1✔
133	subprocStack.add(proc);	1✔
134	}	1✔
135
136	protected synchronized void addSubProcedure(Procedure<Env> proc) {
137	if (!proc.hasParent()) {	1✔
138	return;	1✔
139	}
140	if (subprocs == null) {	1✔
141	subprocs = new HashSet<>();	1✔
142	}
143	subprocs.add(proc);	1✔
144	}	1✔
145
146	/**
147	* Called on store load by the {@link ProcedureExecutor} to load part of the stack.
148	*
149	* <p>Each procedure has its own stack-positions. Which means we have to write to the store only
150	* the Procedure we executed, and nothing else. On load we recreate the full stack by aggregating
151	* each procedure stack-positions.
152	*/
153	protected synchronized void loadStack(Procedure<Env> proc) {
154	addSubProcedure(proc);	1✔
155	int[] stackIndexes = proc.getStackIndexes();	1✔
156	if (stackIndexes != null) {	1✔
157	if (subprocStack == null) {	1✔
158	subprocStack = new ArrayList<>();	1✔
159	}
160	int diff = (1 + stackIndexes[stackIndexes.length - 1]) - subprocStack.size();	1✔
161	if (diff > 0) {	1✔
162	subprocStack.ensureCapacity(1 + stackIndexes[stackIndexes.length - 1]);	1✔
163	while (diff-- > 0) {	1✔
164	subprocStack.add(null);	1✔
165	}
166	}
167	for (int stackIndex : stackIndexes) {	1✔
168	subprocStack.set(stackIndex, proc);	1✔
169	}
170	}
171	if (proc.getState() == ProcedureState.ROLLEDBACK) {	1✔
172	state = State.ROLLINGBACK;	×
173	} else if (proc.isFailed()) {	1✔
174	state = State.FAILED;	×
175	}
176	}	1✔
177	}

apache / iotdb / #9967

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous