Jonathan Lewis über Parallel Execution

Vor ein paar Wochen hat Jonathan Lewis eine Artikelserie begonnen, in der er erklärt, wie parallele Operationen durchgeführt werden und wie man parallele Pläne interpretieren kann:

• Parallel Execution – 1
• weist zunächst darauf hin, dass serielle Pläne in der Regel relativ harmlos bleiben und liefert ein paar Hinweise auf Fälle, in denen das nicht der Fall ist (subquery pushing, Repräsentation von scalar subqueries, Abweichende Reihenfolge für join order und order of operation).
• entscheidend für parallele Ausführung ist, dass möglichst alle Teile des Plans sinnvoll parallelisierbar sind, damit ein Parallelisierungsgrad von N dazu führen kann, dass die Ausführungszeit auf 1/N sinkt. Dabei gibt es vier mögliche Punkte, die diese einfache Rechnung beeinflussen können:
• der setup overhead der Parallelisierung ist nicht kostenlos.
• für parallel query gibt es immer den Schritt der Weitergabe der Daten an das frontend, der eine finale Serialisierung erfordert (und damit ein bottleneck darstellen kann).
• für eine Query mit Parallelisierungsgrad N werden 2 * N slaves verwendet.
• für Exadata ergibt sich eine Parallelisierung auf Ebene der storage server unter Umständen auch für serielle Queries.
• zunächst teilt der query coordinator die Tabelle in N Stücke (bzw. in 13 Stücke, da hier eine - in einem Kommentar des Autors erläuterte - 9:3:1-Strategie eingesetzt wird).
• für Join-Operationen ist es nicht möglich, sicherzustellen, dass die N Teile der einen Tabelle zu den N Teilen der anderen Tabelle passen. Daher werden zwei sets von N parallel exececution slaves eingesetzt, wobei die erste Gruppe die erste Tabelle so effizient wie möglich einliest und die Ergebnisse an die zweite Gruppe übergibt, wobei ein Mechanismus verwendet wird, der einen möglichst effizienten Zugriff auf die zweite Tabelle (bzw. Datenmenge) gewährleisten soll.
• eine Variante ist die Weitergabe der kompletten Ergebnismenge aus der ersten Tabelle an alle slaves der zweiten Gruppe. Diese "(broadcast, none)"-Operation ist dann sinnvoll, wenn nur eine sehr beschränkte Datenmenge aus der ersten Tabelle gelesen wird. Für große Datenmenge werden die Kosten der Vervielfältigung jedoch ein Problem.
• wenn der Zugriff auf die erste Tabelle eine große Ergebnismenge liefert, wird statt der Broadcast-Strategie ein anderes Verfahren verwendet. In diesem Fall wird eine hash Funktion auf die Join-Spalten angewandt, um sicherzustellen, dass die parallel gelesenen Teilbereiche der verknüpften Tabellen zueinander passen. Diese "(hash, hash)"-Operation kann von der Verwendung von Bloom-Filtern profitieren, die eine Filterung vor dem Datentransfer ermöglichen (aber false positives zulassen).
• Parallel Execution – 2
• enthält ein handliches Beispiel mit einem Star Schema bestehend aus einer Fakten- und drei zugehörigen Dimensionstabellen, zu dem eine Query mit einem geeigneten seriellen Plan mit mehreren Hash Joins vorgestellt wird. Dieser Plan wird ausführlich erklärt und besteht im wesentlichen daraus, dass zunächst die Hash Maps der Dimensionstabellen im Speicher aufgebaut und dann die Sätze der Faktentabelle gelesen und gegen die Dimensionsdaten geprüft werden, so dass alle rows für die der probe step erfolgreich verläuft, unmittelbar in der Ergebnismenge landen. Der Hash Join ist in diesem Fall also keine blocking operation.
• Parallel Execution – 3
• liefert den parallelen Plan zur Query, die im vorangehenden Artikel vorgestellt wurde.
• dabei werden einige zusätzliche Hints vorgestellt, die zur expliziten Definition des Parellelisierungsverfahrens verwendet werden können:
• parallel(alias, 2) in Ergänzung zu full(alias): fordert einen parallelen FTS.
  
• pq_distribute(alias none broadcast) in Ergänzung zu use_hash(alias) und swap_join_inputs(alias): definiert das parallelisierte Vorgehen beim Hash Join, in dem es die Methode der Weitergabe der slave sets festlegt.
• es folgt eine detaillierte Analyse des parallelen Zugriffsplans:
• die zu den FTS gehörigen steps PX BLOCK ITERATOR geben an, dass der parallele Scan über rowid ranges erfolgt.
• Paare der Operationen PX SEND / PX RECEIVE zeigen an, dass die parallel slaves ihre Ergebnisse an den nächsten Schritt weitergeben. Die Weitergabe erfolgt dabei über virtuelle Tabellen, deren Namen TQ10000, TQ10001 etc. lauten (wobei TQ für table queue steht).
• die Nummern in den Namen der TQ-Objekte entsprechen der Reihenfolge der Verwendung: TQ10000 wird also als erstes Objekt verwendet, dann TQ10001 usw.
• die TQ-Spalte im Plan gibt an, welche steps für die Füllung einer virtuellen Tabelle verantwortlich sind: Q1,00 verweist dabei auf TQ10000.
• abgesehen von der Parallelisierung entspricht die Verarbeitungsfolge im Beispiel recht exakt der Vorgehensweise der seriellen Ausführung.
• in 12c gibt es weitere Informationen im Plan (etwa zu Bloom-Filtern) und neue Hints (pq_replicate) zur detaillierten Ablaufsteuerung.
• Parallel Execution – 4
• erläutert die in Teil 3 erwähnten Bloom-Filter genauer.
• außerdem werden noch ein paar Details zu den virtual tables und zur parallel execution message size nachgeliefert.
• wichtig ist zunächst, dass ein parallel slave nicht zur gleichen Zeit aus einer virtual table lesen und in eine andere virtual table schreiben kann.
• dabei ist eine virtual table grundsätzlich einfach eine (relativ kleine) Menge von Pages im Speicher, deren Größe durch den Parameter _parallel_execution_message_size bestimmt wird (der unterschiedliche Default-Werte in unterschiedlichen Versionen hat und dessen Größe auch noch von anderen Parametern bestimmt wird, dabei zwischen 2K und 16K liegt).
• Da dieser Speicherbedarf pro communication channel anfällt und jeder slave in einem slave set Daten an jeden slave der anderen slave sets weitergeben kann, ergibt sich in der Summe unter Umständen ein recht hoher Speicherbedarf (errechnet als: * * 3 (= erforderliche Pages) * _parallel_execution_message_size).
• der Speicherplatz wird im shared pool oder in einem large pool "PX msg pool" bereitgestellt.
• die relativ geringe Anzahl verfügbarer Pages in jedem channel führt dazu, dass es zu Wait-Events zwischen lesenden und schreibenden Zugriffen kommt. Diese Events sind: "PX Deq Credit: send blkd" und "PX Deq Credit: need buffer".
• Bloom Filter helfen dabei, das zu transferierende Datenvolumen zu begrenzen und damit die "send blkd" und "need buffer" Waits zu minimieren.
• besonders relevant ist dieser Filterung beim Hash Join (und dabei insbesondere beim Scan der zweiten Tabelle, deren Inhalte mit der in-memory hash table verglichen werden).
• dabei wird der Bloom Filter während des Aufbaus der hash table bestückt und kann dann an die slaves, die den Scan der zweiten Tabelle durchführen, weitergegeben werden. Dabei kann ein Bloom Filter bekanntlich false positives enthalten (was in Christian Antogninis Artikel zum Thema genauer erklärt wird, auf den Jonathan Lewis verweist, und den ich hier auch schon gelegentlich erwähnt habe).
• es folgt ein umfangreiches Beispiel mit Ausführungsplänen. Im Plan erscheint der Bloom Filter als JOIN FILTER CREATE Schritt auf Seite der Erzeugung der hash table und als JOIN FILTER USE beim Scan der zweiten Tabelle. In der predicate section erscheint dabei der Funktionsaufruf sys_op_bloom_filter().
• Parallel Execution – 5
• erläutert die Veränderungen, die sich ergeben, wenn man das Verteilungsverfahren von broadcast auf hash umstellt.
• im Beispiel wird die Änderung des Verfahrens über pq_distribute-Hints hervorgerufen.
• der Plan mit Hash-Verteilung ist deutlich komplizierter als der broadcast-Plan: es erscheinen sieben virtuelle Tabellen (statt vier) und die Verwendung von Bloom Filtern wird intensiver (wobei die sys_op_bloom_filter-Prädikate nicht vollständig sind).
• ergänzend zum Plan mit rowsource-Statistiken werden die zugehörigen Informationen auf v$pq_tqstat aufgeführt.
• Die Analyse folgt wieder der Nummerierung der virtuellen Tabellen. Ich spare mir an dieser Stelle die detaillierte Nacherzählung (die ziemlich ausufernd werden müsste) und erwähne nur, dass die Angaben zu den Bloom-Filtern im Plan nicht notwendig an den richtigen Stellen erscheinen. Zusätzlich gibt's noch eine Grafik zur Visualisierung der Operationen und ihrer Reihenfolge.